Skip to content

feat(enrichers): make domain_to_website extractions opt-out #186

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
rachit367 wants to merge 1 commit into
base: main
Choose a base branch
from
Open

feat(enrichers): make domain_to_website extractions opt-out #186

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
223 changes: 129 additions & 94 deletions flowsint-enrichers/src/flowsint_enrichers/domain/to_website.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,7 +1,6 @@
from typing import Dict, List, Union
from typing import Any, Dict, List
import requests
from bs4 import BeautifulSoup
from flowsint_core.utils import is_valid_domain
from flowsint_core.core.enricher_base import Enricher
from flowsint_enrichers.registry import flowsint_enricher
from flowsint_types.domain import Domain
Expand Down Expand Up @@ -29,8 +28,65 @@ def category(cls) -> str:
def key(cls) -> str:
return "domain"

def _extract_page_info(self, html_content: str) -> Dict[str, any]:
"""Extract title, description, content and technologies from HTML."""
@classmethod
def get_params_schema(cls) -> List[Dict[str, Any]]:
"""Optional toggles to skip the heavier extractions for faster scans."""
return [
{
"name": "extract_content",
"type": "select",
"description": "Extract the page text content (can be slow / large)",
"required": False,
"default": "true",
"options": [
{"label": "Enabled", "value": "true"},
{"label": "Disabled", "value": "false"},
],
},
{
"name": "extract_technologies",
"type": "select",
"description": "Detect web technologies from the HTML",
"required": False,
"default": "true",
"options": [
{"label": "Enabled", "value": "true"},
{"label": "Disabled", "value": "false"},
],
},
{
"name": "extract_headers",
"type": "select",
"description": "Capture relevant HTTP response headers",
"required": False,
"default": "true",
"options": [
{"label": "Enabled", "value": "true"},
{"label": "Disabled", "value": "false"},
],
},
]

def _param_enabled(self, name: str) -> bool:
"""A select param is enabled unless explicitly set to 'false'."""
return self.params.get(name, "true") != "false"

def _extract_headers(self, response: requests.Response) -> Dict[str, str]:
"""Pick out the relevant HTTP headers, dropping missing ones."""
headers = {
"content-type": response.headers.get("content-type"),
"server": response.headers.get("server"),
"x-powered-by": response.headers.get("x-powered-by"),
}
return {k: v for k, v in headers.items() if v}

def _extract_page_info(
self,
html_content: str,
extract_content: bool = True,
extract_technologies: bool = True,
) -> Dict[str, Any]:
"""Extract title, description, and (optionally) content and technologies."""
soup = BeautifulSoup(html_content, 'html.parser')

# Extract title
Expand All @@ -45,46 +101,70 @@ def _extract_page_info(self, html_content: str) -> Dict[str, any]:
if meta_desc and meta_desc.get('content'):
description = meta_desc.get('content').strip()

info: Dict[str, Any] = {'title': title, 'description': description}

# Extract text content (remove scripts and styles)
for script in soup(['script', 'style']):
script.decompose()
content = soup.get_text(separator=' ', strip=True)
# Limit content to first 5000 characters
if len(content) > 5000:
content = content[:5000] + "..."
if extract_content:
for script in soup(['script', 'style']):
script.decompose()
content = soup.get_text(separator=' ', strip=True)
# Limit content to first 5000 characters
if len(content) > 5000:
content = content[:5000] + "..."
info['content'] = content

# Detect technologies
technologies = []

# Check for common frameworks and libraries
if soup.find('meta', attrs={'name': 'generator'}):
generator = soup.find('meta', attrs={'name': 'generator'}).get('content')
if generator:
technologies.append(generator)

# Check for React
if soup.find('div', id='root') or soup.find('div', id='react-root'):
technologies.append('React')

# Check for Vue.js
if soup.find(attrs={'data-v-'}):
technologies.append('Vue.js')

# Check for Angular
if soup.find(attrs={'ng-app'}) or soup.find(attrs={'ng-version'}):
technologies.append('Angular')

# Check for WordPress
if soup.find('meta', attrs={'name': 'generator', 'content': lambda x: x and 'WordPress' in x}):
technologies.append('WordPress')

return {
'title': title,
'description': description,
'content': content,
'technologies': technologies
if extract_technologies:
technologies = []

# Check for common frameworks and libraries
if soup.find('meta', attrs={'name': 'generator'}):
generator = soup.find('meta', attrs={'name': 'generator'}).get('content')
if generator:
technologies.append(generator)

# Check for React
if soup.find('div', id='root') or soup.find('div', id='react-root'):
technologies.append('React')

# Check for Vue.js
if soup.find(attrs={'data-v-'}):
technologies.append('Vue.js')

# Check for Angular
if soup.find(attrs={'ng-app'}) or soup.find(attrs={'ng-version'}):
technologies.append('Angular')

# Check for WordPress
if soup.find('meta', attrs={'name': 'generator', 'content': lambda x: x and 'WordPress' in x}):
technologies.append('WordPress')

info['technologies'] = technologies

return info

def _build_website_data(
self, domain: Domain, url: str, response: requests.Response
) -> Dict[str, Any]:
"""Assemble Website fields from a successful response, honoring params."""
website_data: Dict[str, Any] = {
'url': url,
'domain': domain,
'active': True,
'status_code': response.status_code,
}

if self._param_enabled("extract_headers"):
website_data['headers'] = self._extract_headers(response)

page_info = self._extract_page_info(
response.text,
extract_content=self._param_enabled("extract_content"),
extract_technologies=self._param_enabled("extract_technologies"),
)
website_data.update(page_info)
return website_data

async def scan(self, data: List[InputType]) -> List[OutputType]:
results: List[OutputType] = []
for domain in data:
Expand All @@ -95,65 +175,20 @@ async def scan(self, data: List[InputType]) -> List[OutputType]:
'active': False
}

# Try HTTPS first
try:
https_url = f"https://{domain.domain}"
response = requests.get(
https_url, timeout=10, allow_redirects=True
)

if response.status_code < 400:
website_data['url'] = https_url
website_data['active'] = True
website_data['status_code'] = response.status_code

# Extract relevant headers
website_data['headers'] = {
'content-type': response.headers.get('content-type'),
'server': response.headers.get('server'),
'x-powered-by': response.headers.get('x-powered-by')
}
# Remove None values
website_data['headers'] = {k: v for k, v in website_data['headers'].items() if v}

# Extract page information
page_info = self._extract_page_info(response.text)
website_data.update(page_info)

results.append(Website(**website_data))
# Try HTTPS first, then HTTP.
for scheme in ("https", "http"):
url = f"{scheme}://{domain.domain}"
try:
response = requests.get(
url, timeout=10, allow_redirects=True
)
except requests.RequestException:
continue
except requests.RequestException:
pass

# Try HTTP if HTTPS fails
try:
http_url = f"http://{domain.domain}"
response = requests.get(http_url, timeout=10, allow_redirects=True)

if response.status_code < 400:
website_data['url'] = http_url
website_data['active'] = True
website_data['status_code'] = response.status_code

# Extract relevant headers
website_data['headers'] = {
'content-type': response.headers.get('content-type'),
'server': response.headers.get('server'),
'x-powered-by': response.headers.get('x-powered-by')
}
# Remove None values
website_data['headers'] = {k: v for k, v in website_data['headers'].items() if v}

# Extract page information
page_info = self._extract_page_info(response.text)
website_data.update(page_info)

results.append(Website(**website_data))
continue
except requests.RequestException:
pass
website_data = self._build_website_data(domain, url, response)
break

# If both fail, still add HTTPS URL as default
results.append(Website(**website_data))

except Exception as e:
Expand Down
Loading