diff --git a/flowsint-enrichers/src/flowsint_enrichers/domain/to_phishstats.py b/flowsint-enrichers/src/flowsint_enrichers/domain/to_phishstats.py new file mode 100644 index 000000000..d71659ccf --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/domain/to_phishstats.py @@ -0,0 +1,220 @@ +""" +Domain to PhishStats Enricher + +Enriches domains with phishing intelligence from the PhishStats API. +""" + +from typing import List +from flowsint_core.core.logger import Logger +from flowsint_core.core.enricher_base import Enricher +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.domain import Domain +from flowsint_types.website import Website +from flowsint_types.ip import Ip +from tools.phishstats_client import get_phishstats_client + + +@flowsint_enricher +class DomainToPhishstatsEnricher(Enricher): + """Query PhishStats API for phishing URLs associated with a domain.""" + + InputType = Domain + OutputType = Website + + @classmethod + def name(cls) -> str: + return "domain_to_phishstats" + + @classmethod + def category(cls) -> str: + return "Domain" + + @classmethod + def key(cls) -> str: + return "domain" + + @classmethod + def documentation(cls) -> str: + return """ +# Domain to PhishStats + +Query the PhishStats API to find phishing URLs associated with a domain. + +## What it does + +- Takes domain names as input +- Queries the PhishStats API for all phishing records using that domain +- Returns websites with phishing metadata including: + - URL and page title + - IP address and geographic location + - Security indicators (Safe Browsing, VirusTotal, etc.) + - Statistics (times seen, threat score) + +## Data Source + +- **API**: PhishStats (https://phishstats.info) +- **Rate Limit**: 20 requests per minute +- **Coverage**: Global phishing database updated every 90 minutes + +## Graph Relationships + +Creates the following Neo4j relationships: +- `(Domain)-[:FOUND_IN_PHISHING]->(Website)` +- `(Website)-[:RESOLVES_TO]->(Ip)` +- `(Ip)-[:BELONGS_TO_ASN]->(ASN)` +- `(ASN)-[:OPERATED_BY]->(Organization)` + +## Example + +Input: Domain `github.io` +Output: Website nodes for all phishing sites using that domain +""" + + async def scan(self, data: List[InputType]) -> List[OutputType]: + """Query PhishStats API for each domain.""" + results: List[OutputType] = [] + client = get_phishstats_client() + + for domain_obj in data: + try: + # Query PhishStats API by domain/host - exact match, last 10 results + records = client.query_by_domain_exact(domain_obj.domain, size=10) + + if not records: + Logger.info( + self.sketch_id, + {"message": f"No phishing records found for domain {domain_obj.domain}"} + ) + continue + + # Convert API responses to Website objects + for record in records: + try: + # Create Website with PhishStats metadata + website = Website( + url=record.get("url"), + title=record.get("title"), + status_code=record.get("http_code"), + ) + + # Add PhishStats-specific metadata as extra fields + website.__dict__["phishstats_id"] = record.get("id") + website.__dict__["phishing_score"] = record.get("score") + website.__dict__["date_detected"] = record.get("date") + website.__dict__["google_safebrowsing"] = record.get("google_safebrowsing") + website.__dict__["virus_total"] = record.get("virus_total") + website.__dict__["ip_address"] = record.get("ip") + website.__dict__["asn"] = record.get("asn") + website.__dict__["isp"] = record.get("isp") + website.__dict__["host"] = record.get("host") + website.__dict__["country"] = record.get("countrycode") + + results.append(website) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error parsing PhishStats record: {e}"} + ) + continue + + Logger.info( + self.sketch_id, + {"message": f"Found {len(records)} phishing records for domain {domain_obj.domain}"} + ) + + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error querying PhishStats for domain {domain_obj.domain}: {e}"} + ) + continue + + return results + + def postprocess(self, results: List[OutputType], original_input: List[InputType]) -> List[OutputType]: + """Create graph nodes and relationships.""" + if not self.neo4j_conn: + return results + + # Create nodes for all websites + for website in results: + self.create_node(website) + + # Create relationships between Domains and Websites + for domain_obj in original_input: + self.create_node(domain_obj) + + # Find all Websites that match this domain + matching_sites = [ + w for w in results + if w.__dict__.get("host") and domain_obj.domain in w.__dict__.get("host") + ] + + for website in matching_sites: + self.create_relationship( + domain_obj, + website, + "FOUND_IN_PHISHING" + ) + self.log_graph_message( + f"Domain {domain_obj.domain} found in phishing: {website.url}" + ) + + # Create IP node if IP is present + ip_address = website.__dict__.get("ip_address") + if ip_address: + try: + ip_obj = Ip(address=ip_address) + self.create_node(ip_obj) + self.create_relationship( + website, + ip_obj, + "RESOLVES_TO" + ) + self.log_graph_message( + f"Website resolves to IP {ip_address}" + ) + + # Create ASN node if ASN is present + asn = website.__dict__.get("asn") + if asn: + try: + from flowsint_types.asn import ASN + asn_obj = ASN(asn_str=asn) + self.create_node(asn_obj) + self.create_relationship( + ip_obj, + asn_obj, + "BELONGS_TO_ASN" + ) + self.log_graph_message( + f"IP {ip_address} belongs to ASN {asn}" + ) + + # Create Organization node for ISP + isp = website.__dict__.get("isp") + if isp: + try: + from flowsint_types.organization import Organization + org_obj = Organization(name=isp) + self.create_node(org_obj) + self.create_relationship( + asn_obj, + org_obj, + "OPERATED_BY" + ) + self.log_graph_message( + f"ASN {asn} operated by {isp}" + ) + except Exception: + pass # Invalid organization, skip + except Exception: + pass # Invalid ASN, skip + except Exception: + pass # Invalid IP, skip + + return results + + +InputType = DomainToPhishstatsEnricher.InputType +OutputType = DomainToPhishstatsEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/ip/to_phishstats.py b/flowsint-enrichers/src/flowsint_enrichers/ip/to_phishstats.py new file mode 100644 index 000000000..4f7ec9813 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/ip/to_phishstats.py @@ -0,0 +1,212 @@ +""" +IP to PhishStats Enricher + +Enriches IP addresses with phishing intelligence from the PhishStats API. +""" + +from typing import List +from flowsint_core.core.logger import Logger +from flowsint_core.core.enricher_base import Enricher +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.ip import Ip +from flowsint_types.website import Website +from flowsint_types.domain import Domain +from tools.phishstats_client import get_phishstats_client + + +@flowsint_enricher +class IpToPhishstatsEnricher(Enricher): + """Query PhishStats API for phishing URLs associated with an IP address.""" + + InputType = Ip + OutputType = Website + + @classmethod + def name(cls) -> str: + return "ip_to_phishstats" + + @classmethod + def category(cls) -> str: + return "Ip" + + @classmethod + def key(cls) -> str: + return "address" + + @classmethod + def documentation(cls) -> str: + return """ +# IP to PhishStats + +Query the PhishStats API to find phishing URLs associated with an IP address. + +## What it does + +- Takes IP addresses as input +- Queries the PhishStats API for all phishing records hosted on that IP +- Returns websites with detailed phishing metadata + +## Data Source + +- **API**: PhishStats (https://phishstats.info) +- **Rate Limit**: 20 requests per minute +- **Coverage**: Global phishing database updated every 90 minutes + +## Graph Relationships + +Creates the following Neo4j relationships: +- `(Ip)-[:HOSTS_PHISHING]->(Website)` +- `(Website)-[:USES_DOMAIN]->(Domain)` +- `(Ip)-[:BELONGS_TO_ASN]->(ASN)` +- `(ASN)-[:OPERATED_BY]->(Organization)` + +## Example + +Input: IP `185.199.110.153` +Output: Website nodes for all phishing URLs hosted on that IP +""" + + async def scan(self, data: List[InputType]) -> List[OutputType]: + """Query PhishStats API for each IP address.""" + results: List[OutputType] = [] + client = get_phishstats_client() + + for ip_obj in data: + try: + # Query PhishStats API - exact match, last 10 results + records = client.query_by_ip(ip_obj.address, size=10) + + if not records: + Logger.info( + self.sketch_id, + {"message": f"No phishing records found for IP {ip_obj.address}"} + ) + continue + + # Convert API responses to Website objects + for record in records: + try: + website = Website( + url=record.get("url"), + title=record.get("title"), + status_code=record.get("http_code"), + ) + + # Add PhishStats metadata + website.__dict__["phishstats_id"] = record.get("id") + website.__dict__["phishing_score"] = record.get("score") + website.__dict__["date_detected"] = record.get("date") + website.__dict__["google_safebrowsing"] = record.get("google_safebrowsing") + website.__dict__["virus_total"] = record.get("virus_total") + website.__dict__["ip_address"] = record.get("ip") + website.__dict__["asn"] = record.get("asn") + website.__dict__["isp"] = record.get("isp") + website.__dict__["host"] = record.get("host") + website.__dict__["country"] = record.get("countrycode") + + results.append(website) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error parsing PhishStats record: {e}"} + ) + continue + + Logger.info( + self.sketch_id, + {"message": f"Found {len(records)} phishing records for IP {ip_obj.address}"} + ) + + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error querying PhishStats for IP {ip_obj.address}: {e}"} + ) + continue + + return results + + def postprocess(self, results: List[OutputType], original_input: List[InputType]) -> List[OutputType]: + """Create graph nodes and relationships.""" + if not self.neo4j_conn: + return results + + # Create nodes for all websites + for website in results: + self.create_node(website) + + # Create relationships between IPs and Websites + for ip_obj in original_input: + self.create_node(ip_obj) + + # Find all Websites that match this IP + matching_sites = [w for w in results if w.__dict__.get("ip_address") == ip_obj.address] + + for website in matching_sites: + self.create_relationship( + ip_obj, + website, + "HOSTS_PHISHING" + ) + self.log_graph_message( + f"IP {ip_obj.address} hosts phishing: {website.url}" + ) + + # Create Domain node if host is present + host = website.__dict__.get("host") + if host: + try: + domain_obj = Domain(domain=host) + self.create_node(domain_obj) + self.create_relationship( + website, + domain_obj, + "USES_DOMAIN" + ) + self.log_graph_message( + f"Website uses domain {host}" + ) + except Exception: + pass # Invalid domain, skip + + # Create ASN node if ASN is present + asn = website.__dict__.get("asn") + if asn: + try: + from flowsint_types.asn import ASN + asn_obj = ASN(asn_str=asn) + self.create_node(asn_obj) + self.create_relationship( + ip_obj, + asn_obj, + "BELONGS_TO_ASN" + ) + self.log_graph_message( + f"IP {ip_obj.address} belongs to ASN {asn}" + ) + + # Create Organization node for ISP + isp = website.__dict__.get("isp") + if isp: + try: + from flowsint_types.organization import Organization + org_obj = Organization(name=isp) + self.create_node(org_obj) + self.create_relationship( + asn_obj, + org_obj, + "OPERATED_BY" + ) + self.log_graph_message( + f"ASN {asn} operated by {isp}" + ) + except Exception: + pass # Invalid organization, skip + except Exception: + pass # Invalid ASN, skip + + return results + + +InputType = IpToPhishstatsEnricher.InputType +OutputType = IpToPhishstatsEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/phrase/to_phishstats.py b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_phishstats.py new file mode 100644 index 000000000..2eb7b3f08 --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/phrase/to_phishstats.py @@ -0,0 +1,268 @@ +""" +Phrase to PhishStats Enricher + +Search PhishStats for URLs and titles matching a phrase pattern. +""" + +from typing import List +from flowsint_core.core.logger import Logger +from flowsint_core.core.enricher_base import Enricher +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.phrase import Phrase +from flowsint_types.website import Website +from flowsint_types.domain import Domain +from flowsint_types.ip import Ip +from tools.phishstats_client import get_phishstats_client + + +@flowsint_enricher +class PhraseToPhishstatsEnricher(Enricher): + """Search PhishStats for phishing URLs matching a phrase in title or URL.""" + + InputType = Phrase + OutputType = Website + + @classmethod + def name(cls) -> str: + return "phrase_to_phishstats" + + @classmethod + def category(cls) -> str: + return "Phrase" + + @classmethod + def key(cls) -> str: + return "text" + + @classmethod + def documentation(cls) -> str: + return """ +# Phrase to PhishStats + +Search the PhishStats API for phishing URLs where the phrase appears in the title OR URL. + +## What it does + +- Takes a phrase/keyword as input +- Searches PhishStats for matches in: + - Page titles (like,~phrase~) + - URLs (like,~phrase~) +- Returns matching phishing websites with full metadata +- Results sorted by newest first + +## Data Source + +- **API**: PhishStats (https://phishstats.info) +- **Rate Limit**: 20 requests per minute +- **Coverage**: Global phishing database updated every 90 minutes + +## Query Pattern + +Uses OR logic to search both fields: +``` +_where=(title,like,~{phrase}~)~or(url,like,~{phrase}~) +``` + +## Graph Relationships + +Creates the following Neo4j relationships: +- `(Phrase)-[:FOUND_IN_PHISHING]->(Website)` +- `(Website)-[:USES_DOMAIN]->(Domain)` +- `(Website)-[:RESOLVES_TO]->(Ip)` +- `(Ip)-[:BELONGS_TO_ASN]->(ASN)` +- `(ASN)-[:OPERATED_BY]->(Organization)` + +## Examples + +**Search for "facebook" phishing:** +- Input: Phrase "facebook" +- Finds: Sites with "facebook" in title or URL + +**Search for "paypal" phishing:** +- Input: Phrase "paypal" +- Finds: Sites with "paypal" in title or URL + +**Search for bank names:** +- Input: Phrase "chase" +- Finds: Phishing sites targeting Chase bank +""" + + async def scan(self, data: List[InputType]) -> List[OutputType]: + """Search PhishStats for each phrase.""" + results: List[OutputType] = [] + client = get_phishstats_client() + + for phrase_obj in data: + try: + phrase_text = str(phrase_obj.text) + + # Build OR query: (title,like,~phrase~)~or(url,like,~phrase~) + where_clause = f"(title,like,~{phrase_text}~)~or(url,like,~{phrase_text}~)" + + Logger.info( + self.sketch_id, + {"message": f"Searching PhishStats for phrase: {phrase_text}"} + ) + + # Query with the OR clause, last 10 results + records = client.query(where_clause, size=10, sort="-id") + + if not records: + Logger.info( + self.sketch_id, + {"message": f"No phishing records found for phrase '{phrase_text}'"} + ) + continue + + # Convert API responses to Website objects + for record in records: + try: + # Create Website with PhishStats metadata + website = Website( + url=record.get("url"), + title=record.get("title"), + status_code=record.get("http_code"), + ) + + # Add PhishStats-specific metadata as extra fields + # These will be stored as properties on the node + website.__dict__["phishstats_id"] = record.get("id") + website.__dict__["phishing_score"] = record.get("score") + website.__dict__["date_detected"] = record.get("date") + website.__dict__["google_safebrowsing"] = record.get("google_safebrowsing") + website.__dict__["virus_total"] = record.get("virus_total") + website.__dict__["ip_address"] = record.get("ip") + website.__dict__["asn"] = record.get("asn") + website.__dict__["isp"] = record.get("isp") + website.__dict__["host"] = record.get("host") + website.__dict__["country"] = record.get("countrycode") + + results.append(website) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error parsing PhishStats record: {e}"} + ) + continue + + Logger.info( + self.sketch_id, + {"message": f"Found {len(records)} phishing records for phrase'{phrase_text}'"} + ) + + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error querying PhishStats for phrase '{phrase_obj.text}': {e}"} + ) + continue + + return results + + def postprocess(self, results: List[OutputType], original_input: List[InputType]) -> List[OutputType]: + """Create graph nodes and relationships.""" + if not self.neo4j_conn: + return results + + # Create nodes for all websites + for website in results: + self.create_node(website) + + # Create relationships between Phrases and Websites + for phrase_obj in original_input: + self.create_node(phrase_obj) + + phrase_text = str(phrase_obj.text).lower() + + # Find all Websites that match this phrase + matching_sites = [ + w for w in results + if phrase_text in str(w.url).lower() or + (w.title and phrase_text in w.title.lower()) + ] + + for website in matching_sites: + self.create_relationship( + phrase_obj, + website, + "FOUND_IN_PHISHING" + ) + self.log_graph_message( + f"Phrase '{phrase_obj.text}' found in phishing: {website.url}" + ) + + # Create Domain node if host is present + host = website.__dict__.get("host") + if host: + try: + domain_obj = Domain(domain=host) + self.create_node(domain_obj) + self.create_relationship( + website, + domain_obj, + "USES_DOMAIN" + ) + self.log_graph_message( + f"Website uses domain {host}" + ) + except Exception: + pass # Invalid domain, skip + + # Create IP node if IP is present + ip_address = website.__dict__.get("ip_address") + if ip_address: + try: + ip_obj = Ip(address=ip_address) + self.create_node(ip_obj) + self.create_relationship( + website, + ip_obj, + "RESOLVES_TO" + ) + self.log_graph_message( + f"Website resolves to IP {ip_address}" + ) + + # Create ASN node if ASN is present + asn = website.__dict__.get("asn") + if asn: + try: + from flowsint_types.asn import ASN + asn_obj = ASN(asn_str=asn) + self.create_node(asn_obj) + self.create_relationship( + ip_obj, + asn_obj, + "BELONGS_TO_ASN" + ) + self.log_graph_message( + f"IP {ip_address} belongs to ASN {asn}" + ) + + # Create Organization node for ISP + isp = website.__dict__.get("isp") + if isp: + try: + from flowsint_types.organization import Organization + org_obj = Organization(name=isp) + self.create_node(org_obj) + self.create_relationship( + asn_obj, + org_obj, + "OPERATED_BY" + ) + self.log_graph_message( + f"ASN {asn} operated by {isp}" + ) + except Exception: + pass # Invalid organization, skip + except Exception: + pass # Invalid ASN, skip + except Exception: + pass # Invalid IP, skip + + return results + + +InputType = PhraseToPhishstatsEnricher.InputType +OutputType = PhraseToPhishstatsEnricher.OutputType diff --git a/flowsint-enrichers/src/flowsint_enrichers/website/to_phishstats.py b/flowsint-enrichers/src/flowsint_enrichers/website/to_phishstats.py new file mode 100644 index 000000000..91186a1ba --- /dev/null +++ b/flowsint-enrichers/src/flowsint_enrichers/website/to_phishstats.py @@ -0,0 +1,238 @@ +""" +Website to PhishStats Enricher + +Enriches websites/URLs with phishing intelligence from the PhishStats API. +""" + +from typing import List +from flowsint_core.core.logger import Logger +from flowsint_core.core.enricher_base import Enricher +from flowsint_enrichers.registry import flowsint_enricher +from flowsint_types.website import Website +from flowsint_types.domain import Domain +from flowsint_types.ip import Ip +from tools.phishstats_client import get_phishstats_client + + +@flowsint_enricher +class WebsiteToPhishstatsEnricher(Enricher): + """Query PhishStats API for phishing records matching a website URL.""" + + InputType = Website + OutputType = Website + + @classmethod + def name(cls) -> str: + return "website_to_phishstats" + + @classmethod + def category(cls) -> str: + return "Website" + + @classmethod + def key(cls) -> str: + return "url" + + @classmethod + def documentation(cls) -> str: + return """ +# Website to PhishStats + +Query the PhishStats API to find phishing records matching a website URL. + +## What it does + +- Takes Website URLs as input +- Queries the PhishStats API for exact URL matches +- Returns website with PhishStats metadata if found + +## Data Source + +- **API**: PhishStats (https://phishstats.info) +- **Rate Limit**: 20 requests per minute +- **Coverage**: Global phishing database updated every 90 minutes + +## Graph Relationships + +Creates the following Neo4j relationships: +- `(Website)-[:FOUND_IN_PHISHING]->(Website)` (enriched with metadata) +- `(Website)-[:USES_DOMAIN]->(Domain)` +- `(Website)-[:RESOLVES_TO]->(Ip)` +- `(Ip)-[:BELONGS_TO_ASN]->(ASN)` +- `(ASN)-[:OPERATED_BY]->(Organization)` + +## Example + +Input: Website `https://github.io/phishing` +Output: Same website enriched with PhishStats metadata (if found) +""" + + async def scan(self, data: List[InputType]) -> List[OutputType]: + """Query PhishStats API for each website URL.""" + results: List[OutputType] = [] + client = get_phishstats_client() + + for website_obj in data: + try: + # Extract the URL string + url_str = str(website_obj.url) + + # Query PhishStats API by URL - exact match, last 1 result (URLs are unique) + records = client.query_by_url_exact(url_str, size=1) + + if not records: + Logger.info( + self.sketch_id, + {"message": f"No phishing records found for URL {url_str}"} + ) + continue + + # Convert API responses to Website objects + for record in records: + try: + website = Website( + url=record.get("url"), + title=record.get("title"), + status_code=record.get("http_code"), + ) + + # Add PhishStats metadata + website.__dict__["phishstats_id"] = record.get("id") + website.__dict__["phishing_score"] = record.get("score") + website.__dict__["date_detected"] = record.get("date") + website.__dict__["google_safebrowsing"] = record.get("google_safebrowsing") + website.__dict__["virus_total"] = record.get("virus_total") + website.__dict__["ip_address"] = record.get("ip") + website.__dict__["asn"] = record.get("asn") + website.__dict__["isp"] = record.get("isp") + website.__dict__["host"] = record.get("host") + website.__dict__["country"] = record.get("countrycode") + + results.append(website) + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error parsing PhishStats record: {e}"} + ) + continue + + Logger.info( + self.sketch_id, + {"message": f"Found {len(records)} phishing records for URL {url_str}"} + ) + + except Exception as e: + Logger.error( + self.sketch_id, + {"message": f"Error querying PhishStats for URL {website_obj.url}: {e}"} + ) + continue + + return results + + def postprocess(self, results: List[OutputType], original_input: List[InputType]) -> List[OutputType]: + """Create graph nodes and relationships.""" + if not self.neo4j_conn: + return results + + # Create nodes for all websites + for website in results: + self.create_node(website) + + # Create relationships between input Websites and found phishing Websites + for input_website in original_input: + self.create_node(input_website) + + url_str = str(input_website.url).lower() + + # Find matching websites + matching_sites = [ + w for w in results + if url_str in str(w.url).lower() or str(w.url).lower() in url_str + ] + + for website in matching_sites: + self.create_relationship( + input_website, + website, + "FOUND_IN_PHISHING" + ) + self.log_graph_message( + f"Website {input_website.url} found in phishing database" + ) + + # Create Domain node if host is present + host = website.__dict__.get("host") + if host: + try: + domain_obj = Domain(domain=host) + self.create_node(domain_obj) + self.create_relationship( + website, + domain_obj, + "USES_DOMAIN" + ) + self.log_graph_message( + f"Website uses domain {host}" + ) + except Exception: + pass # Invalid domain, skip + + # Create IP node if IP is present + ip_address = website.__dict__.get("ip_address") + if ip_address: + try: + ip_obj = Ip(address=ip_address) + self.create_node(ip_obj) + self.create_relationship( + website, + ip_obj, + "RESOLVES_TO" + ) + self.log_graph_message( + f"Website resolves to IP {ip_address}" + ) + + # Create ASN node if ASN is present + asn = website.__dict__.get("asn") + if asn: + try: + from flowsint_types.asn import ASN + asn_obj = ASN(asn_str=asn) + self.create_node(asn_obj) + self.create_relationship( + ip_obj, + asn_obj, + "BELONGS_TO_ASN" + ) + self.log_graph_message( + f"IP {ip_address} belongs to ASN {asn}" + ) + + # Create Organization node for ISP + isp = website.__dict__.get("isp") + if isp: + try: + from flowsint_types.organization import Organization + org_obj = Organization(name=isp) + self.create_node(org_obj) + self.create_relationship( + asn_obj, + org_obj, + "OPERATED_BY" + ) + self.log_graph_message( + f"ASN {asn} operated by {isp}" + ) + except Exception: + pass # Invalid organization, skip + except Exception: + pass # Invalid ASN, skip + except Exception: + pass # Invalid IP, skip + + return results + + +InputType = WebsiteToPhishstatsEnricher.InputType +OutputType = WebsiteToPhishstatsEnricher.OutputType diff --git a/flowsint-enrichers/src/tools/phishstats_client.py b/flowsint-enrichers/src/tools/phishstats_client.py new file mode 100644 index 000000000..36092eeba --- /dev/null +++ b/flowsint-enrichers/src/tools/phishstats_client.py @@ -0,0 +1,246 @@ +""" +PhishStats API Client + +A reusable client for interacting with the PhishStats API, handling +rate limiting, error handling, and response parsing. +""" + +import time +import requests +from typing import Dict, List, Optional, Any + + +class PhishStatsClient: + """Client for querying the PhishStats API with rate limiting and error handling.""" + + BASE_URL = "https://api.phishstats.info/api/phishing" + RATE_LIMIT_PER_MINUTE = 20 + DEFAULT_TIMEOUT = 60 # Increased from 20 to 60 seconds + + def __init__(self): + """Initialize the PhishStats API client.""" + self.request_times: List[float] = [] + self.session = requests.Session() + self.session.headers.update({ + "User-Agent": "flowsint-phishstats-enricher/1.0" + }) + + def _check_rate_limit(self): + """ + Check and enforce rate limiting (20 requests per minute). + Sleeps if necessary to stay within limits. + """ + now = time.time() + # Remove requests older than 60 seconds + self.request_times = [t for t in self.request_times if now - t < 60] + + if len(self.request_times) >= self.RATE_LIMIT_PER_MINUTE: + # Calculate how long to wait + oldest_request = self.request_times[0] + wait_time = 60 - (now - oldest_request) + if wait_time > 0: + print(f"Rate limit reached. Waiting {wait_time:.2f} seconds...") + time.sleep(wait_time + 0.1) # Add small buffer + # Clean up old timestamps after sleeping + now = time.time() + self.request_times = [t for t in self.request_times if now - t < 60] + + # Record this request + self.request_times.append(time.time()) + + def _make_request( + self, + endpoint: str = "", + params: Optional[Dict[str, Any]] = None, + retries: int = 2 # Reduced from 3 to 2 + ) -> Any: + """ + Make a request to the PhishStats API with retries and error handling. + + Args: + endpoint: API endpoint (e.g., "", "count") + params: Query parameters + retries: Number of retry attempts + + Returns: + Parsed JSON response + + Raises: + Exception: If request fails after all retries + """ + url = f"{self.BASE_URL}/{endpoint}" if endpoint else self.BASE_URL + + for attempt in range(retries): + try: + # Enforce rate limiting + self._check_rate_limit() + + # Make the request + print(f"Making request to PhishStats API (attempt {attempt + 1}/{retries})...") + response = self.session.get( + url, + params=params, + timeout=self.DEFAULT_TIMEOUT + ) + + # Handle rate limiting + if response.status_code == 429: + wait_time = (2 ** attempt) * 5 # Exponential backoff + print(f"Rate limited (429). Waiting {wait_time} seconds before retry...") + time.sleep(wait_time) + continue + + # Raise for other HTTP errors + response.raise_for_status() + + # Parse and return JSON + data = response.json() + print(f"Successfully received {len(data) if isinstance(data, list) else 1} records") + return data + + except requests.exceptions.Timeout: + if attempt < retries - 1: + print(f"Request timeout after {self.DEFAULT_TIMEOUT}s. Retrying ({attempt + 1}/{retries})...") + time.sleep(2 ** attempt) + else: + raise Exception(f"Request timed out after {retries} attempts (timeout: {self.DEFAULT_TIMEOUT}s each)") + + except requests.exceptions.RequestException as e: + if attempt < retries - 1: + print(f"Request failed: {e}. Retrying ({attempt + 1}/{retries})...") + time.sleep(2 ** attempt) + else: + raise Exception(f"Request failed after {retries} attempts: {e}") + + raise Exception("Request failed after all retries") + + def query( + self, + where_clause: str, + size: int = 100, + sort: str = "-id", + fields: Optional[str] = None, + page: int = 1 + ) -> List[Dict[str, Any]]: + """ + Query the PhishStats API. + + Args: + where_clause: WHERE clause in PhishStats format, e.g., "(ip,eq,1.2.3.4)" + size: Number of results to return (max 100) + sort: Sort field, prefix with "-" for descending + fields: Comma-separated list of fields to return + page: Page number for pagination + + Returns: + List of phishing record dictionaries + """ + params = { + "_where": where_clause, + "_sort": sort, + "_size": min(size, 100), # Cap at 100 + "_p": page + } + + if fields: + params["_fields"] = fields + + return self._make_request(params=params) + + def count(self, where_clause: str) -> int: + """ + Get the count of records matching the criteria. + + Args: + where_clause: WHERE clause in PhishStats format + + Returns: + Count of matching records + """ + params = {"_where": where_clause} + result = self._make_request(endpoint="count", params=params) + + if result and isinstance(result, list) and len(result) > 0: + return result[0].get("no_of_rows", 0) + return 0 + + def query_by_ip(self, ip: str, size: int = 100) -> List[Dict[str, Any]]: + """Query phishing records by IP address (exact match).""" + where_clause = f"(ip,eq,{ip})" + return self.query(where_clause, size=size) + + def query_by_domain(self, domain: str, size: int = 100) -> List[Dict[str, Any]]: + """Query phishing records by domain/host (using LIKE for flexibility).""" + where_clause = f"(host,like,~{domain}~)" + return self.query(where_clause, size=size) + + def query_by_domain_exact(self, domain: str, size: int = 100) -> List[Dict[str, Any]]: + """Query phishing records by domain/host (exact match).""" + where_clause = f"(host,eq,{domain})" + return self.query(where_clause, size=size) + + def query_by_url(self, url: str, size: int = 100) -> List[Dict[str, Any]]: + """Query phishing records by URL pattern (using LIKE).""" + where_clause = f"(url,like,~{url}~)" + return self.query(where_clause, size=size) + + def query_by_url_exact(self, url: str, size: int = 100) -> List[Dict[str, Any]]: + """Query phishing records by URL (exact match).""" + where_clause = f"(url,eq,{url})" + return self.query(where_clause, size=size) + + def search( + self, + field: str, + operator: str, + value: str, + size: int = 100 + ) -> List[Dict[str, Any]]: + """ + Flexible search method. + + Args: + field: Field name (e.g., "title", "url", "ip") + operator: Operator (e.g., "eq", "like", "gt") + value: Search value (use ~value~ for LIKE) + size: Number of results + + Returns: + List of matching records + """ + where_clause = f"({field},{operator},{value})" + return self.query(where_clause, size=size) + + def build_where_clause(self, conditions: List[tuple], logic: str = "and") -> str: + """ + Build a complex WHERE clause from multiple conditions. + + Args: + conditions: List of tuples (field, operator, value) + logic: Logical operator ("and" or "or") + + Returns: + WHERE clause string + + Example: + build_where_clause([("ip", "eq", "1.2.3.4"), ("score", "gt", "5")]) + Returns: "(ip,eq,1.2.3.4)~and(score,gt,5)" + """ + if not conditions: + return "" + + clauses = [f"({field},{op},{value})" for field, op, value in conditions] + logic_op = f"~{logic}~" + return logic_op.join(clauses) + + +# Singleton instance for reuse across enrichers +_client_instance = None + + +def get_phishstats_client() -> PhishStatsClient: + """Get or create a singleton PhishStats client instance.""" + global _client_instance + if _client_instance is None: + _client_instance = PhishStatsClient() + return _client_instance