mirror of https://github.com/CIRCL/lookyloo
1344 lines
65 KiB
Python
1344 lines
65 KiB
Python
#!/usr/bin/env python3
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import copy
|
|
import gzip
|
|
import json
|
|
import logging
|
|
import operator
|
|
import smtplib
|
|
import ssl
|
|
import time
|
|
|
|
from collections import defaultdict
|
|
from datetime import date, datetime, timedelta
|
|
from email.message import EmailMessage
|
|
from functools import cached_property
|
|
from io import BytesIO
|
|
from pathlib import Path
|
|
from typing import Any, Iterable, TYPE_CHECKING, overload, Literal
|
|
from urllib.parse import urlparse
|
|
from uuid import uuid4
|
|
from zipfile import ZipFile
|
|
|
|
import mmh3
|
|
|
|
from defang import defang # type: ignore[import-untyped]
|
|
from har2tree import CrawledTree, HostNode, URLNode
|
|
from lacuscore import (LacusCore,
|
|
CaptureStatus as CaptureStatusCore,
|
|
# CaptureResponse as CaptureResponseCore)
|
|
# CaptureResponseJson as CaptureResponseJsonCore,
|
|
CaptureSettings as CaptureSettingsCore)
|
|
from PIL import Image, UnidentifiedImageError
|
|
from playwrightcapture import get_devices
|
|
from puremagic import from_string, PureError # type: ignore[import-untyped]
|
|
from pylacus import (PyLacus,
|
|
CaptureStatus as CaptureStatusPy
|
|
# CaptureResponse as CaptureResponsePy,
|
|
# CaptureResponseJson as CaptureResponseJsonPy,
|
|
# CaptureSettings as CaptureSettingsPy
|
|
)
|
|
from pymisp import MISPAttribute, MISPEvent, MISPObject
|
|
from pysecuritytxt import PySecurityTXT, SecurityTXTNotAvailable
|
|
from pylookyloomonitoring import PyLookylooMonitoring # type: ignore[attr-defined]
|
|
from redis import ConnectionPool, Redis
|
|
from redis.connection import UnixDomainSocketConnection
|
|
|
|
from .capturecache import CaptureCache, CapturesIndex
|
|
from .context import Context
|
|
from .default import LookylooException, get_homedir, get_config, get_socket_path, safe_create_dir
|
|
from .exceptions import (MissingCaptureDirectory,
|
|
MissingUUID, TreeNeedsRebuild, NoValidHarFile)
|
|
from .helpers import (get_captures_dir, get_email_template,
|
|
get_resources_hashes, get_taxonomies,
|
|
uniq_domains, ParsedUserAgent, load_cookies, UserAgents,
|
|
get_useragent_for_requests, make_ts_from_dirname)
|
|
from .modules import (MISPs, PhishingInitiative, UniversalWhois,
|
|
UrlScan, VirusTotal, Phishtank, Hashlookup,
|
|
RiskIQ, RiskIQError, Pandora, URLhaus, CIRCLPDNS)
|
|
|
|
if TYPE_CHECKING:
|
|
from playwright.async_api import Cookie
|
|
|
|
|
|
class CaptureSettings(CaptureSettingsCore, total=False):
|
|
'''The capture settings that can be passed to Lookyloo'''
|
|
listing: int | None
|
|
not_queued: int | None
|
|
auto_report: bool | str | dict[str, str] | None
|
|
dnt: str | None
|
|
browser_name: str | None
|
|
os: str | None
|
|
parent: str | None
|
|
|
|
|
|
class Lookyloo():
|
|
|
|
def __init__(self, cache_max_size: int | None=None) -> None:
|
|
'''Initialize lookyloo.
|
|
:param cache_max_size: The maximum size of the cache. Alows to display captures metadata without getting it from redis
|
|
This cache is *not* useful for background indexing or pickle building, only for the front end.
|
|
So it should always be None *unless* we're running the background processes.
|
|
'''
|
|
self.logger = logging.getLogger(f'{self.__class__.__name__}')
|
|
self.logger.setLevel(get_config('generic', 'loglevel'))
|
|
self.user_agents = UserAgents()
|
|
self.is_public_instance = get_config('generic', 'public_instance')
|
|
self.public_domain = get_config('generic', 'public_domain')
|
|
|
|
self.global_proxy = {}
|
|
if global_proxy := get_config('generic', 'global_proxy'):
|
|
if global_proxy.get('enable'):
|
|
self.global_proxy = copy.copy(global_proxy)
|
|
self.global_proxy.pop('enable')
|
|
|
|
self.securitytxt = PySecurityTXT(useragent=get_useragent_for_requests())
|
|
self.taxonomies = get_taxonomies()
|
|
|
|
self.redis_pool: ConnectionPool = ConnectionPool(connection_class=UnixDomainSocketConnection,
|
|
path=get_socket_path('cache'), decode_responses=True)
|
|
self.capture_dir: Path = get_captures_dir()
|
|
|
|
self._priority = get_config('generic', 'priority')
|
|
|
|
# Initialize 3rd party components
|
|
# ## Initialize MISP(s)
|
|
try_old_config = False
|
|
# New config
|
|
self.misps = MISPs(config_name='MultipleMISPs')
|
|
if not self.misps.available:
|
|
self.logger.warning('Unable to setup the MISPs module')
|
|
try_old_config = True
|
|
|
|
if try_old_config:
|
|
# Legacy MISP config, now use MultipleMISPs key to support more than one MISP instance
|
|
try:
|
|
if misp_config := get_config('modules', 'MISP'):
|
|
misps_config = {'default': 'MISP', 'instances': {'MISP': misp_config}}
|
|
self.misps = MISPs(config=misps_config)
|
|
if self.misps.available:
|
|
self.logger.warning('Please migrate the MISP config to the "MultipleMISPs" key in the config, and remove the "MISP" key')
|
|
else:
|
|
self.logger.warning('Unable to setup the MISP module')
|
|
except Exception:
|
|
# The key was removed from the config, and the sample config
|
|
pass
|
|
|
|
# ## Done with MISP(s)
|
|
|
|
self.pi = PhishingInitiative(config_name='PhishingInitiative')
|
|
self.vt = VirusTotal(config_name='VirusTotal')
|
|
self.uwhois = UniversalWhois(config_name='UniversalWhois')
|
|
self.urlscan = UrlScan(config_name='UrlScan')
|
|
self.phishtank = Phishtank(config_name='Phishtank')
|
|
self.hashlookup = Hashlookup(config_name='Hashlookup')
|
|
self.riskiq = RiskIQ(config_name='RiskIQ')
|
|
self.pandora = Pandora(config_name='Pandora')
|
|
self.urlhaus = URLhaus(config_name='URLhaus')
|
|
self.circl_pdns = CIRCLPDNS(config_name='CIRCLPDNS')
|
|
|
|
self.monitoring_enabled = False
|
|
if monitoring_config := get_config('generic', 'monitoring'):
|
|
if monitoring_config['enable']:
|
|
self.monitoring = PyLookylooMonitoring(monitoring_config['url'], get_useragent_for_requests())
|
|
if self.monitoring.is_up:
|
|
self.monitoring_enabled = True
|
|
# NOTE: maybe move that somewhere else: we'll need to restart the webserver
|
|
# if we change the settings in the monitoring instance
|
|
self.monitoring_settings = self.monitoring.instance_settings()
|
|
|
|
self.logger.info('Initializing context...')
|
|
self.context = Context()
|
|
self.logger.info('Context initialized.')
|
|
self.logger.info('Initializing index...')
|
|
self._captures_index = CapturesIndex(self.redis, self.context, maxsize=cache_max_size)
|
|
self.logger.info('Index initialized.')
|
|
|
|
# init lacus
|
|
self.lacus
|
|
|
|
@property
|
|
def redis(self) -> Redis: # type: ignore[type-arg]
|
|
return Redis(connection_pool=self.redis_pool)
|
|
|
|
@cached_property
|
|
def lacus(self) -> PyLacus | LacusCore:
|
|
has_remote_lacus = False
|
|
self._lacus: PyLacus | LacusCore
|
|
if get_config('generic', 'remote_lacus'):
|
|
remote_lacus_config = get_config('generic', 'remote_lacus')
|
|
if remote_lacus_config.get('enable'):
|
|
self.logger.info("Remote lacus enabled, trying to set it up...")
|
|
lacus_retries = 10
|
|
while lacus_retries > 0:
|
|
remote_lacus_url = remote_lacus_config.get('url')
|
|
self._lacus = PyLacus(remote_lacus_url)
|
|
if self._lacus.is_up:
|
|
has_remote_lacus = True
|
|
self.logger.info(f"Remote lacus enabled to {remote_lacus_url}.")
|
|
break
|
|
lacus_retries -= 1
|
|
self.logger.warning(f"Unable to setup remote lacus to {remote_lacus_url}, trying again {lacus_retries} more time(s).")
|
|
time.sleep(10)
|
|
else:
|
|
raise LookylooException('Remote lacus is enabled but unreachable.')
|
|
|
|
if not has_remote_lacus:
|
|
# We need a redis connector that doesn't decode.
|
|
redis: Redis = Redis(unix_socket_path=get_socket_path('cache')) # type: ignore[type-arg]
|
|
self._lacus = LacusCore(redis, tor_proxy=get_config('generic', 'tor_proxy'),
|
|
max_capture_time=get_config('generic', 'max_capture_time'),
|
|
only_global_lookups=get_config('generic', 'only_global_lookups'),
|
|
loglevel=get_config('generic', 'loglevel'))
|
|
return self._lacus
|
|
|
|
def add_context(self, capture_uuid: str, /, urlnode_uuid: str, *, ressource_hash: str,
|
|
legitimate: bool, malicious: bool, details: dict[str, dict[str, str]]) -> None:
|
|
'''Adds context information to a capture or a URL node'''
|
|
if malicious:
|
|
self.context.add_malicious(ressource_hash, details['malicious'])
|
|
if legitimate:
|
|
self.context.add_legitimate(ressource_hash, details['legitimate'])
|
|
|
|
def add_to_legitimate(self, capture_uuid: str, /, hostnode_uuid: str | None=None, urlnode_uuid: str | None=None) -> None:
|
|
'''Mark a full capture as legitimate.
|
|
Iterates over all the nodes and mark them all as legitimate too.'''
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
self.context.mark_as_legitimate(ct, hostnode_uuid, urlnode_uuid)
|
|
|
|
def remove_pickle(self, capture_uuid: str, /) -> None:
|
|
'''Remove the pickle from a specific capture.'''
|
|
self._captures_index.remove_pickle(capture_uuid)
|
|
|
|
def rebuild_cache(self) -> None:
|
|
'''Flush and rebuild the redis cache. Doesn't remove the pickles.
|
|
The cached captures will be rebuild when loading the index.'''
|
|
self.redis.flushdb()
|
|
|
|
def rebuild_all(self) -> None:
|
|
'''Flush and rebuild the redis cache, and delete all the pickles.
|
|
The captures will be rebuilt by the background indexer'''
|
|
self._captures_index.rebuild_all()
|
|
|
|
def get_urlnode_from_tree(self, capture_uuid: str, /, node_uuid: str) -> URLNode:
|
|
'''Get a URL node from a tree, by UUID'''
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
return ct.root_hartree.get_url_node_by_uuid(node_uuid)
|
|
|
|
def get_hostnode_from_tree(self, capture_uuid: str, /, node_uuid: str) -> HostNode:
|
|
'''Get a host node from a tree, by UUID'''
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
return ct.root_hartree.get_host_node_by_uuid(node_uuid)
|
|
|
|
def get_statistics(self, capture_uuid: str, /) -> dict[str, Any]:
|
|
'''Get the statistics of a capture.'''
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
return ct.root_hartree.stats
|
|
|
|
def get_info(self, capture_uuid: str, /) -> dict[str, Any]:
|
|
'''Get basic information about the capture.'''
|
|
cache = self.capture_cache(capture_uuid)
|
|
if not cache:
|
|
return {'error': f'Unable to find UUID {capture_uuid} in the cache.'}
|
|
|
|
if not hasattr(cache, 'uuid'):
|
|
self.logger.critical(f'Cache for {capture_uuid} is broken: {cache}.')
|
|
return {'error': f'Sorry, the capture {capture_uuid} is broken, please report it to the admin.'}
|
|
|
|
to_return = {'uuid': cache.uuid,
|
|
'url': cache.url if hasattr(cache, 'url') else 'Unable to get URL for the capture'}
|
|
if hasattr(cache, 'error') and cache.error:
|
|
to_return['error'] = cache.error
|
|
if hasattr(cache, 'title'):
|
|
to_return['title'] = cache.title
|
|
if hasattr(cache, 'timestamp'):
|
|
to_return['capture_time'] = cache.timestamp.isoformat()
|
|
if hasattr(cache, 'user_agent') and cache.user_agent:
|
|
to_return['user_agent'] = cache.user_agent
|
|
if hasattr(cache, 'referer'):
|
|
to_return['referer'] = cache.referer if cache.referer else ''
|
|
return to_return
|
|
|
|
def get_meta(self, capture_uuid: str, /) -> dict[str, str]:
|
|
'''Get the meta informations from a capture (mostly, details about the User Agent used.)'''
|
|
cache = self.capture_cache(capture_uuid)
|
|
if not cache:
|
|
return {}
|
|
metafile = cache.capture_dir / 'meta'
|
|
if metafile.exists():
|
|
with metafile.open('r') as f:
|
|
return json.load(f)
|
|
|
|
if not cache.user_agent:
|
|
return {}
|
|
meta = {}
|
|
ua = ParsedUserAgent(cache.user_agent)
|
|
meta['user_agent'] = ua.string
|
|
if ua.platform:
|
|
meta['os'] = ua.platform
|
|
if ua.browser:
|
|
if ua.version:
|
|
meta['browser'] = f'{ua.browser} {ua.version}'
|
|
else:
|
|
meta['browser'] = ua.browser
|
|
|
|
if not meta:
|
|
# UA not recognized
|
|
self.logger.info(f'Unable to recognize the User agent: {ua}')
|
|
with metafile.open('w') as f:
|
|
json.dump(meta, f)
|
|
return meta
|
|
|
|
def get_capture_settings(self, capture_uuid: str, /) -> CaptureSettings:
|
|
cache = self.capture_cache(capture_uuid)
|
|
if not cache:
|
|
return {}
|
|
cs_file = cache.capture_dir / 'capture_settings.json'
|
|
if cs_file.exists():
|
|
with cs_file.open('r') as f:
|
|
return json.load(f)
|
|
return {}
|
|
|
|
def categories_capture(self, capture_uuid: str, /) -> dict[str, Any]:
|
|
'''Get all the categories related to a capture, in MISP Taxonomies format'''
|
|
categ_file = self._captures_index[capture_uuid].capture_dir / 'categories'
|
|
# get existing categories if possible
|
|
if categ_file.exists():
|
|
with categ_file.open() as f:
|
|
current_categories = [line.strip() for line in f.readlines()]
|
|
return {e: self.taxonomies.revert_machinetag(e) for e in current_categories}
|
|
return {}
|
|
|
|
def categorize_capture(self, capture_uuid: str, /, category: str) -> None:
|
|
'''Add a category (MISP Taxonomy tag) to a capture.'''
|
|
if not get_config('generic', 'enable_categorization'):
|
|
return
|
|
# Make sure the category is mappable to a taxonomy.
|
|
self.taxonomies.revert_machinetag(category)
|
|
|
|
categ_file = self._captures_index[capture_uuid].capture_dir / 'categories'
|
|
# get existing categories if possible
|
|
if categ_file.exists():
|
|
with categ_file.open() as f:
|
|
current_categories = {line.strip() for line in f.readlines()}
|
|
else:
|
|
current_categories = set()
|
|
current_categories.add(category)
|
|
with categ_file.open('w') as f:
|
|
f.writelines(f'{t}\n' for t in current_categories)
|
|
|
|
def uncategorize_capture(self, capture_uuid: str, /, category: str) -> None:
|
|
'''Remove a category (MISP Taxonomy tag) from a capture.'''
|
|
if not get_config('generic', 'enable_categorization'):
|
|
return
|
|
categ_file = self._captures_index[capture_uuid].capture_dir / 'categories'
|
|
# get existing categories if possible
|
|
if categ_file.exists():
|
|
with categ_file.open() as f:
|
|
current_categories = {line.strip() for line in f.readlines()}
|
|
else:
|
|
current_categories = set()
|
|
current_categories.remove(category)
|
|
with categ_file.open('w') as f:
|
|
f.writelines(f'{t}\n' for t in current_categories)
|
|
|
|
def trigger_modules(self, capture_uuid: str, /, force: bool=False, auto_trigger: bool=False) -> dict[str, Any]:
|
|
'''Launch the 3rd party modules on a capture.
|
|
It uses the cached result *if* the module was triggered the same day.
|
|
The `force` flag re-triggers the module regardless of the cache.'''
|
|
try:
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
except LookylooException:
|
|
self.logger.warning(f'Unable to trigger the modules unless the tree ({capture_uuid}) is cached.')
|
|
return {'error': f'UUID {capture_uuid} is either unknown or the tree is not ready yet.'}
|
|
|
|
self.uwhois.capture_default_trigger(ct, force=force, auto_trigger=auto_trigger)
|
|
self.hashlookup.capture_default_trigger(ct, auto_trigger=auto_trigger)
|
|
|
|
to_return: dict[str, dict[str, Any]] = {'PhishingInitiative': {}, 'VirusTotal': {}, 'UrlScan': {},
|
|
'URLhaus': {}}
|
|
if cache := self.capture_cache(capture_uuid):
|
|
to_return['PhishingInitiative'] = self.pi.capture_default_trigger(cache, force=force, auto_trigger=auto_trigger)
|
|
to_return['VirusTotal'] = self.vt.capture_default_trigger(cache, force=force, auto_trigger=auto_trigger)
|
|
to_return['UrlScan'] = self.urlscan.capture_default_trigger(
|
|
cache,
|
|
visibility='unlisted' if (cache and cache.no_index) else 'public',
|
|
force=force, auto_trigger=auto_trigger)
|
|
to_return['Phishtank'] = self.phishtank.capture_default_trigger(cache, auto_trigger=auto_trigger)
|
|
to_return['URLhaus'] = self.urlhaus.capture_default_trigger(cache, auto_trigger=auto_trigger)
|
|
return to_return
|
|
|
|
def get_modules_responses(self, capture_uuid: str, /) -> dict[str, Any]:
|
|
'''Get the responses of the modules from the cached responses on the disk'''
|
|
cache = self.capture_cache(capture_uuid)
|
|
# TODO: return a message when we cannot get the modules responses, update the code checking if it is falsy accordingly.
|
|
if not cache:
|
|
self.logger.warning(f'Unable to get the modules responses unless the capture {capture_uuid} is cached')
|
|
return {}
|
|
if not hasattr(cache, 'url'):
|
|
self.logger.warning(f'The capture {capture_uuid} does not have a URL in the cache, it is broken.')
|
|
return {}
|
|
|
|
to_return: dict[str, Any] = {}
|
|
if self.vt.available:
|
|
to_return['vt'] = {}
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
for redirect in cache.redirects:
|
|
to_return['vt'][redirect] = self.vt.get_url_lookup(redirect)
|
|
else:
|
|
to_return['vt'][cache.url] = self.vt.get_url_lookup(cache.url)
|
|
if self.pi.available:
|
|
to_return['pi'] = {}
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
for redirect in cache.redirects:
|
|
to_return['pi'][redirect] = self.pi.get_url_lookup(redirect)
|
|
else:
|
|
to_return['pi'][cache.url] = self.pi.get_url_lookup(cache.url)
|
|
if self.phishtank.available:
|
|
to_return['phishtank'] = {'urls': {}, 'ips_hits': {}}
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
for redirect in cache.redirects:
|
|
to_return['phishtank']['urls'][redirect] = self.phishtank.get_url_lookup(redirect)
|
|
else:
|
|
to_return['phishtank']['urls'][cache.url] = self.phishtank.get_url_lookup(cache.url)
|
|
ips_hits = self.phishtank.lookup_ips_capture(cache)
|
|
if ips_hits:
|
|
to_return['phishtank']['ips_hits'] = ips_hits
|
|
if self.urlhaus.available:
|
|
to_return['urlhaus'] = {'urls': {}}
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
for redirect in cache.redirects:
|
|
to_return['urlhaus']['urls'][redirect] = self.urlhaus.get_url_lookup(redirect)
|
|
else:
|
|
to_return['urlhaus']['urls'][cache.url] = self.urlhaus.get_url_lookup(cache.url)
|
|
|
|
if self.urlscan.available:
|
|
to_return['urlscan'] = {'submission': {}, 'result': {}}
|
|
to_return['urlscan']['submission'] = self.urlscan.get_url_submission(cache)
|
|
if to_return['urlscan']['submission'] and 'uuid' in to_return['urlscan']['submission']:
|
|
# The submission was done, try to get the results
|
|
result = self.urlscan.url_result(cache)
|
|
if 'error' not in result:
|
|
to_return['urlscan']['result'] = result
|
|
return to_return
|
|
|
|
def get_historical_lookups(self, capture_uuid: str, /, force: bool=False) -> dict[str, Any]:
|
|
# this method is only trigered when the user wants to get more details about the capture
|
|
# by looking at Passive DNS systems, check if there are hits in the current capture
|
|
# in another one and things like that. The trigger_modules method is for getting
|
|
# information about the current status of the capture in other systems.
|
|
cache = self.capture_cache(capture_uuid)
|
|
if not cache:
|
|
self.logger.warning(f'Unable to get the modules responses unless the capture {capture_uuid} is cached')
|
|
return {}
|
|
to_return: dict[str, Any] = defaultdict(dict)
|
|
if self.riskiq.available:
|
|
try:
|
|
self.riskiq.capture_default_trigger(cache)
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
hostname = urlparse(cache.redirects[-1]).hostname
|
|
else:
|
|
hostname = urlparse(cache.url).hostname
|
|
if hostname:
|
|
if _riskiq_entries := self.riskiq.get_passivedns(hostname):
|
|
to_return['riskiq'] = _riskiq_entries
|
|
except RiskIQError as e:
|
|
self.logger.warning(e.response.content)
|
|
if self.circl_pdns.available:
|
|
self.circl_pdns.capture_default_trigger(cache)
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
hostname = urlparse(cache.redirects[-1]).hostname
|
|
else:
|
|
hostname = urlparse(cache.url).hostname
|
|
if hostname:
|
|
if _circl_entries := self.circl_pdns.get_passivedns(hostname):
|
|
to_return['circl_pdns'][hostname] = _circl_entries
|
|
return to_return
|
|
|
|
def hide_capture(self, capture_uuid: str, /) -> None:
|
|
"""Add the capture in the hidden pool (not shown on the front page)
|
|
NOTE: it won't remove the correlations until they are rebuilt.
|
|
"""
|
|
capture_dir = self._captures_index[capture_uuid].capture_dir
|
|
self.redis.hset(str(capture_dir), 'no_index', 1)
|
|
(capture_dir / 'no_index').touch()
|
|
self._captures_index.reload_cache(capture_uuid)
|
|
|
|
def update_tree_cache_info(self, process_id: int, classname: str) -> None:
|
|
self.redis.hset('tree_cache', f'{process_id}|{classname}', str(self._captures_index.lru_cache_status()))
|
|
|
|
def clear_tree_cache(self) -> None:
|
|
self._captures_index.lru_cache_clear()
|
|
|
|
def sorted_capture_cache(self, capture_uuids: Iterable[str] | None=None, cached_captures_only: bool=True, index_cut_time: datetime | None=None) -> list[CaptureCache]:
|
|
'''Get all the captures in the cache, sorted by timestamp (new -> old).
|
|
By default, this method will only return the captures that are currently cached.'''
|
|
# Make sure we do not try to load archived captures that would still be in 'lookup_dirs'
|
|
cut_time = (datetime.now() - timedelta(days=get_config('generic', 'archive') - 1))
|
|
if index_cut_time:
|
|
if index_cut_time < cut_time:
|
|
index_cut_time = cut_time
|
|
else:
|
|
index_cut_time = cut_time
|
|
|
|
if capture_uuids is None:
|
|
capture_uuids = {uuid for uuid, directory in self.redis.hscan_iter('lookup_dirs')
|
|
if make_ts_from_dirname(directory.rsplit('/', 1)[-1]) > index_cut_time}
|
|
# NOTE: we absolutely have to respect the cached_captures_only setting and
|
|
# never overwrite it. This method is called to display the index
|
|
# and if we try to display everything, including the non-cached entries,
|
|
# the index can get stuck building a lot of captures
|
|
# cached_captures_only = False
|
|
|
|
if not capture_uuids:
|
|
# No captures at all on the instance
|
|
return []
|
|
|
|
if cached_captures_only:
|
|
# Do not try to build pickles
|
|
capture_uuids = set(capture_uuids) & self._captures_index.cached_captures
|
|
|
|
all_cache: list[CaptureCache] = [self._captures_index[uuid] for uuid in capture_uuids
|
|
if self.capture_cache(uuid)
|
|
and hasattr(self._captures_index[uuid], 'timestamp')]
|
|
all_cache.sort(key=operator.attrgetter('timestamp'), reverse=True)
|
|
return all_cache
|
|
|
|
def get_capture_status(self, capture_uuid: str, /) -> CaptureStatusCore | CaptureStatusPy:
|
|
'''Returns the status (queued, ongoing, done, or UUID unknown)'''
|
|
if self.redis.hexists('lookup_dirs', capture_uuid):
|
|
return CaptureStatusCore.DONE
|
|
elif self.redis.sismember('ongoing', capture_uuid):
|
|
# Post-processing on lookyloo's side
|
|
return CaptureStatusCore.ONGOING
|
|
try:
|
|
lacus_status = self.lacus.get_capture_status(capture_uuid)
|
|
except Exception as e:
|
|
self.logger.warning(f'Unable to get the status for {capture_uuid} from lacus: {e}')
|
|
if self.redis.zscore('to_capture', capture_uuid) is not None:
|
|
return CaptureStatusCore.QUEUED
|
|
else:
|
|
return CaptureStatusCore.UNKNOWN
|
|
|
|
if (lacus_status == CaptureStatusCore.UNKNOWN
|
|
and self.redis.zscore('to_capture', capture_uuid) is not None):
|
|
# If we do the query before lacus picks it up, we will tell to the user that the UUID doesn't exists.
|
|
return CaptureStatusCore.QUEUED
|
|
elif lacus_status == CaptureStatusCore.DONE:
|
|
# Done on lacus side, but not processed by Lookyloo yet (it would be in lookup_dirs)
|
|
return CaptureStatusCore.ONGOING
|
|
return lacus_status
|
|
|
|
def capture_cache(self, capture_uuid: str, /, *, force_update: bool = False) -> CaptureCache | None:
|
|
"""Get the cache from redis, rebuild the tree if the internal UUID changed => slow"""
|
|
try:
|
|
cache = self._captures_index[capture_uuid]
|
|
if cache and force_update:
|
|
needs_update = False
|
|
if not cache.user_agent and not cache.error:
|
|
# 2022-12-07: New cache format, store the user agent and referers.
|
|
needs_update = True
|
|
if not hasattr(cache, 'title') or not cache.title:
|
|
# 2023-17-27: The title should *always* be there,
|
|
# unless the HAR file is missing or broken
|
|
needs_update = True
|
|
if needs_update:
|
|
self._captures_index.reload_cache(capture_uuid)
|
|
cache = self._captures_index[capture_uuid]
|
|
return cache
|
|
except NoValidHarFile:
|
|
self.logger.debug('No HAR files, {capture_uuid} is a broken capture.')
|
|
return None
|
|
except MissingCaptureDirectory as e:
|
|
# The UUID is in the captures but the directory is not on the disk.
|
|
self.logger.warning(f'Missing Directory: {e}')
|
|
return None
|
|
except MissingUUID:
|
|
if self.get_capture_status(capture_uuid) not in [CaptureStatusCore.QUEUED, CaptureStatusCore.ONGOING]:
|
|
self.logger.info(f'Unable to find {capture_uuid} (not in the cache and/or missing capture directory).')
|
|
return None
|
|
except LookylooException as e:
|
|
self.logger.warning(f'Lookyloo Exception: {e}')
|
|
return None
|
|
except Exception as e:
|
|
self.logger.exception(e)
|
|
return None
|
|
|
|
def get_crawled_tree(self, capture_uuid: str, /) -> CrawledTree:
|
|
'''Get the generated tree in ETE Toolkit format.
|
|
Loads the pickle if it exists, creates it otherwise.'''
|
|
try:
|
|
return self._captures_index[capture_uuid].tree
|
|
except TreeNeedsRebuild:
|
|
self._captures_index.reload_cache(capture_uuid)
|
|
return self._captures_index[capture_uuid].tree
|
|
|
|
def _prepare_lacus_query(self, query: CaptureSettings) -> CaptureSettings:
|
|
# Remove the none, it makes redis unhappy
|
|
query = {k: v for k, v in query.items() if v is not None} # type: ignore[assignment]
|
|
|
|
if 'url' in query and query['url'] is not None:
|
|
# Make sure the URL does not have any space or newline
|
|
query['url'] = query['url'].strip()
|
|
|
|
# NOTE: Lookyloo' capture can pass a do not track header independently from the default headers, merging it here
|
|
headers = query.pop('headers', {})
|
|
if 'dnt' in query:
|
|
if isinstance(headers, str):
|
|
headers += f'\nDNT: {query.pop("dnt")}'
|
|
headers = headers.strip()
|
|
elif isinstance(headers, dict):
|
|
dnt_entry = query.pop("dnt")
|
|
if dnt_entry:
|
|
headers['DNT'] = dnt_entry.strip()
|
|
|
|
if headers:
|
|
query['headers'] = headers
|
|
|
|
# NOTE: Lookyloo can get the cookies in somewhat weird formats, mornalizing them
|
|
query['cookies'] = load_cookies(query.pop('cookies', None))
|
|
|
|
# NOTE: Make sure we have a useragent
|
|
user_agent = query.pop('user_agent', None)
|
|
if not user_agent:
|
|
# Catch case where the UA is broken on the UI, and the async submission.
|
|
self.user_agents.user_agents # triggers an update of the default UAs
|
|
if 'device_name' not in query:
|
|
query['user_agent'] = user_agent if user_agent else self.user_agents.default['useragent']
|
|
|
|
# NOTE: the document must be base64 encoded
|
|
document: str | bytes | None = query.pop('document', None)
|
|
if document:
|
|
if isinstance(document, bytes):
|
|
query['document'] = base64.b64encode(document).decode()
|
|
else:
|
|
query['document'] = document
|
|
return query
|
|
|
|
def enqueue_capture(self, query: CaptureSettings, source: str, user: str, authenticated: bool) -> str:
|
|
'''Enqueue a query in the capture queue (used by the UI and the API for asynchronous processing)'''
|
|
|
|
def get_priority(source: str, user: str, authenticated: bool) -> int:
|
|
src_prio: int = self._priority['sources'][source] if source in self._priority['sources'] else -1
|
|
if not authenticated:
|
|
usr_prio = self._priority['users']['_default_anon']
|
|
# reduce priority for anonymous users making lots of captures
|
|
queue_size = self.redis.zscore('queues', f'{source}|{authenticated}|{user}')
|
|
if queue_size is None:
|
|
queue_size = 0
|
|
usr_prio -= int(queue_size / 10)
|
|
else:
|
|
usr_prio = self._priority['users'][user] if self._priority['users'].get(user) else self._priority['users']['_default_auth']
|
|
return src_prio + usr_prio
|
|
|
|
for key, value in query.items():
|
|
if isinstance(value, bool):
|
|
query[key] = 1 if value else 0 # type: ignore[literal-required]
|
|
elif isinstance(value, (list, dict)):
|
|
query[key] = json.dumps(value) if value else None # type: ignore[literal-required]
|
|
|
|
query = self._prepare_lacus_query(query)
|
|
|
|
priority = get_priority(source, user, authenticated)
|
|
if priority < -100:
|
|
# Someone is probably abusing the system with useless URLs, remove them from the index
|
|
query['listing'] = 0
|
|
try:
|
|
perma_uuid = self.lacus.enqueue( # type: ignore[misc]
|
|
url=query.get('url', None),
|
|
document_name=query.get('document_name', None),
|
|
document=query.get('document', None),
|
|
# depth=query.get('depth', 0),
|
|
browser=query.get('browser', None), # type: ignore[arg-type]
|
|
device_name=query.get('device_name', None),
|
|
user_agent=query.get('user_agent', None),
|
|
proxy=self.global_proxy if self.global_proxy else query.get('proxy', None),
|
|
general_timeout_in_sec=query.get('general_timeout_in_sec', None),
|
|
cookies=query.get('cookies', None),
|
|
headers=query.get('headers', None),
|
|
http_credentials=query.get('http_credentials', None),
|
|
viewport=query.get('viewport', None),
|
|
referer=query.get('referer', None),
|
|
timezone_id=query.get('timezone_id', None),
|
|
locale=query.get('locale', None),
|
|
geolocation=query.get('geolocation', None),
|
|
color_scheme=query.get('color_scheme', None),
|
|
rendered_hostname_only=query.get('rendered_hostname_only', True),
|
|
with_favicon=query.get('with_favicon', True),
|
|
allow_tracking=query.get('allow_tracking', True),
|
|
# force=query.get('force', False),
|
|
# recapture_interval=query.get('recapture_interval', 300),
|
|
priority=priority
|
|
)
|
|
except Exception as e:
|
|
self.logger.critical(f'Unable to enqueue capture: {e}')
|
|
perma_uuid = str(uuid4())
|
|
query['not_queued'] = 1
|
|
finally:
|
|
if (not self.redis.hexists('lookup_dirs', perma_uuid) # already captured
|
|
and self.redis.zscore('to_capture', perma_uuid) is None): # capture ongoing
|
|
|
|
# Make the settings redis compatible
|
|
mapping_capture: dict[str, bytes | float | int | str] = {}
|
|
for key, value in query.items():
|
|
if isinstance(value, bool):
|
|
mapping_capture[key] = 1 if value else 0
|
|
elif isinstance(value, (list, dict)):
|
|
if value:
|
|
mapping_capture[key] = json.dumps(value)
|
|
elif value is not None:
|
|
mapping_capture[key] = value # type: ignore[assignment]
|
|
|
|
p = self.redis.pipeline()
|
|
p.zadd('to_capture', {perma_uuid: priority})
|
|
p.hset(perma_uuid, mapping=mapping_capture) # type: ignore[arg-type]
|
|
p.zincrby('queues', 1, f'{source}|{authenticated}|{user}')
|
|
p.set(f'{perma_uuid}_mgmt', f'{source}|{authenticated}|{user}')
|
|
p.execute()
|
|
|
|
return perma_uuid
|
|
|
|
def takedown_details(self, hostnode: HostNode) -> dict[str, Any]:
|
|
if not self.uwhois.available:
|
|
self.logger.warning('UWhois module not enabled, unable to use this method')
|
|
raise LookylooException('UWhois module not enabled, unable to use this method')
|
|
to_return = {'hostname': hostnode.name,
|
|
'contacts': self.uwhois.whois(hostnode.name, contact_email_only=True), # List of emails from whois
|
|
'ips': {}, # ip: [list of contacts from whois]
|
|
'asns': {}, # ASN: [list of contacts from whois]
|
|
'all_emails': set()
|
|
}
|
|
to_return['ips'] = {ip: self.uwhois.whois(ip, contact_email_only=True) for ip in set(hostnode.resolved_ips['v4']) | set(hostnode.resolved_ips['v6'])}
|
|
to_return['asns'] = {asn['asn']: self.uwhois.whois(f'AS{asn["asn"]}', contact_email_only=True) for asn in hostnode.ipasn.values()}
|
|
|
|
# try to get contact from security.txt file
|
|
try:
|
|
txtfile = self.securitytxt.get(hostnode.name)
|
|
parsed = self.securitytxt.parse(txtfile)
|
|
to_return['securitytxt'] = parsed
|
|
if 'contact' in parsed:
|
|
if isinstance(parsed['contact'], str):
|
|
to_return['all_emails'].add(parsed['contact'].lstrip('mailto:'))
|
|
else:
|
|
to_return['all_emails'] |= {contact.lstrip('mailto:') for contact in parsed['contact'] if contact.startswith('mailto:')}
|
|
except SecurityTXTNotAvailable as e:
|
|
self.logger.debug(f'Unable to get a security.txt file: {e}')
|
|
|
|
for emails in to_return['ips'].values():
|
|
to_return['all_emails'] |= set(emails)
|
|
|
|
for emails in to_return['asns'].values():
|
|
to_return['all_emails'] |= set(emails)
|
|
|
|
# URLs specific details
|
|
|
|
# # IPFS
|
|
for url in hostnode.urls:
|
|
for h in url.response['headers']:
|
|
if h['name'].lower().startswith('x-ipfs'):
|
|
# got an ipfs thing
|
|
to_return['all_emails'].add('abuse@ipfs.io')
|
|
if 'urls' not in to_return:
|
|
to_return['urls'] = {'ipfs': {}}
|
|
if url.name not in to_return['urls']['ipfs']:
|
|
to_return['urls']['ipfs'][url.name] = ['abuse@ipfs.io']
|
|
else:
|
|
to_return['urls']['ipfs'][url.name].append('abuse@ipfs.io')
|
|
break
|
|
|
|
to_return['all_emails'] = list(to_return['all_emails'])
|
|
return to_return
|
|
|
|
def contacts(self, capture_uuid: str, /) -> list[dict[str, Any]]:
|
|
capture = self.get_crawled_tree(capture_uuid)
|
|
rendered_hostnode = self.get_hostnode_from_tree(capture_uuid, capture.root_hartree.rendered_node.hostnode_uuid)
|
|
result = []
|
|
for node in reversed(rendered_hostnode.get_ancestors()):
|
|
result.append(self.takedown_details(node))
|
|
result.append(self.takedown_details(rendered_hostnode))
|
|
return result
|
|
|
|
def send_mail(self, capture_uuid: str, /, email: str='', comment: str | None=None) -> bool | dict[str, Any]:
|
|
'''Send an email notification regarding a specific capture'''
|
|
if not get_config('generic', 'enable_mail_notification'):
|
|
return {"error": "Unable to send mail: mail notification disabled"}
|
|
|
|
email_config = get_config('generic', 'email')
|
|
smtp_auth = get_config('generic', 'email_smtp_auth')
|
|
redirects = ''
|
|
initial_url = ''
|
|
if cache := self.capture_cache(capture_uuid):
|
|
if hasattr(cache, 'url'):
|
|
if email_config['defang_urls']:
|
|
initial_url = defang(cache.url, colon=True, all_dots=True)
|
|
else:
|
|
initial_url = cache.url
|
|
else:
|
|
initial_url = 'Unable to get URL from cache, this is probably a bug.'
|
|
if hasattr(cache, 'error') and cache.error:
|
|
initial_url += f' - {cache.error}'
|
|
|
|
if hasattr(cache, 'redirects') and cache.redirects:
|
|
redirects = "Redirects:\n"
|
|
if email_config['defang_urls']:
|
|
redirects += defang('\n'.join(cache.redirects), colon=True, all_dots=True)
|
|
else:
|
|
redirects += '\n'.join(cache.redirects)
|
|
else:
|
|
redirects = "No redirects."
|
|
|
|
msg = EmailMessage()
|
|
msg['From'] = email_config['from']
|
|
if email:
|
|
msg['Reply-To'] = email
|
|
msg['To'] = email_config['to']
|
|
msg['Subject'] = email_config['subject']
|
|
body = get_email_template()
|
|
body = body.format(
|
|
recipient=msg['To'].addresses[0].display_name,
|
|
domain=self.public_domain,
|
|
uuid=capture_uuid,
|
|
initial_url=initial_url,
|
|
redirects=redirects,
|
|
comment=comment if comment else '',
|
|
sender=msg['From'].addresses[0].display_name,
|
|
)
|
|
msg.set_content(body)
|
|
try:
|
|
contact_for_takedown = self.contacts(capture_uuid)
|
|
msg.add_attachment(json.dumps(contact_for_takedown, indent=2), filename='contacts.json')
|
|
except Exception as e:
|
|
self.logger.warning(f'Unable to get the contacts: {e}')
|
|
try:
|
|
with smtplib.SMTP(email_config['smtp_host'], email_config['smtp_port']) as s:
|
|
if smtp_auth['auth']:
|
|
if smtp_auth['smtp_use_starttls']:
|
|
if smtp_auth['verify_certificate'] is False:
|
|
ssl_context = ssl.create_default_context()
|
|
ssl_context.check_hostname = False
|
|
ssl_context.verify_mode = ssl.CERT_NONE
|
|
s.starttls(context=ssl_context)
|
|
else:
|
|
s.starttls()
|
|
s.login(smtp_auth['smtp_user'], smtp_auth['smtp_pass'])
|
|
s.send_message(msg)
|
|
except Exception as e:
|
|
self.logger.exception(e)
|
|
self.logger.warning(msg.as_string())
|
|
return {"error": "Unable to send mail"}
|
|
return True
|
|
|
|
def _get_raw(self, capture_uuid: str, /, extension: str='*', all_files: bool=True) -> BytesIO:
|
|
'''Get file(s) from the capture directory'''
|
|
try:
|
|
capture_dir = self._captures_index[capture_uuid].capture_dir
|
|
except NoValidHarFile:
|
|
return BytesIO(f'Capture {capture_uuid} has no HAR entries, which means it is broken.'.encode())
|
|
except MissingUUID:
|
|
return BytesIO(f'Capture {capture_uuid} not unavailable, try again later.'.encode())
|
|
except MissingCaptureDirectory:
|
|
return BytesIO(f'No capture {capture_uuid} on the system (directory missing).'.encode())
|
|
all_paths = sorted(list(capture_dir.glob(f'*.{extension}')))
|
|
if not all_files:
|
|
# Only get the first one in the list
|
|
if not all_paths:
|
|
return BytesIO()
|
|
with open(all_paths[0], 'rb') as f:
|
|
return BytesIO(f.read())
|
|
to_return = BytesIO()
|
|
# Add uuid file to the export, allows to keep the same UUID across platforms.
|
|
# NOTE: the UUID file will always be added, as long as all_files is True,
|
|
# even if we pass an extension
|
|
all_paths.append(capture_dir / 'uuid')
|
|
with ZipFile(to_return, 'w') as myzip:
|
|
for path in all_paths:
|
|
if 'pickle' in path.name:
|
|
# We do not want to export the pickle
|
|
continue
|
|
myzip.write(path, arcname=f'{capture_dir.name}/{path.name}')
|
|
to_return.seek(0)
|
|
return to_return
|
|
|
|
@overload
|
|
def get_potential_favicons(self, capture_uuid: str, /, all_favicons: Literal[False], for_datauri: Literal[True]) -> tuple[str, str]:
|
|
...
|
|
|
|
@overload
|
|
def get_potential_favicons(self, capture_uuid: str, /, all_favicons: Literal[True], for_datauri: Literal[False]) -> BytesIO:
|
|
...
|
|
|
|
def get_potential_favicons(self, capture_uuid: str, /, all_favicons: bool=False, for_datauri: bool=False) -> BytesIO | tuple[str, str]:
|
|
'''Get rendered HTML'''
|
|
fav = self._get_raw(capture_uuid, 'potential_favicons.ico', all_favicons)
|
|
if not all_favicons and for_datauri:
|
|
favicon = fav.getvalue()
|
|
if not favicon:
|
|
return '', ''
|
|
try:
|
|
mimetype = from_string(favicon, mime=True)
|
|
return mimetype, base64.b64encode(favicon).decode()
|
|
except PureError:
|
|
self.logger.warning(f'Unable to get the mimetype of the favicon for {capture_uuid}.')
|
|
return '', ''
|
|
return fav
|
|
|
|
def get_html(self, capture_uuid: str, /, all_html: bool=False) -> BytesIO:
|
|
'''Get rendered HTML'''
|
|
return self._get_raw(capture_uuid, 'html', all_html)
|
|
|
|
def get_data(self, capture_uuid: str, /) -> tuple[str, BytesIO]:
|
|
'''Get the data'''
|
|
return self._get_raw(capture_uuid, 'data.filename', False).getvalue().decode(), self._get_raw(capture_uuid, 'data', False)
|
|
|
|
def get_cookies(self, capture_uuid: str, /, all_cookies: bool=False) -> BytesIO:
|
|
'''Get the cookie(s)'''
|
|
return self._get_raw(capture_uuid, 'cookies.json', all_cookies)
|
|
|
|
def get_screenshot(self, capture_uuid: str, /) -> BytesIO:
|
|
'''Get the screenshot(s) of the rendered page'''
|
|
return self._get_raw(capture_uuid, 'png', all_files=False)
|
|
|
|
def get_screenshot_thumbnail(self, capture_uuid: str, /, for_datauri: bool=False, width: int=64) -> str | BytesIO:
|
|
'''Get the thumbnail of the rendered page. Always crop to a square.'''
|
|
to_return = BytesIO()
|
|
size = width, width
|
|
try:
|
|
s = self.get_screenshot(capture_uuid)
|
|
orig_screenshot = Image.open(s)
|
|
to_thumbnail = orig_screenshot.crop((0, 0, orig_screenshot.width, orig_screenshot.width))
|
|
except Image.DecompressionBombError as e:
|
|
# The image is most probably too big: https://pillow.readthedocs.io/en/stable/reference/Image.html
|
|
self.logger.warning(f'Unable to generate the screenshot thumbnail of {capture_uuid}: image too big ({e}).')
|
|
error_img: Path = get_homedir() / 'website' / 'web' / 'static' / 'error_screenshot.png'
|
|
to_thumbnail = Image.open(error_img)
|
|
except UnidentifiedImageError as e:
|
|
# We might have a direct download link, and no screenshot. Assign the thumbnail accordingly.
|
|
try:
|
|
filename, data = self.get_data(capture_uuid)
|
|
if filename:
|
|
self.logger.info(f'{capture_uuid} is is a download link, set thumbnail.')
|
|
error_img = get_homedir() / 'website' / 'web' / 'static' / 'download.png'
|
|
else:
|
|
# No screenshot and no data, it is probably because the capture failed.
|
|
error_img = get_homedir() / 'website' / 'web' / 'static' / 'error_screenshot.png'
|
|
except Exception:
|
|
# The capture probably doesn't have a screenshot at all, no need to log that as a warning.
|
|
self.logger.debug(f'Unable to generate the screenshot thumbnail of {capture_uuid}: {e}.')
|
|
error_img = get_homedir() / 'website' / 'web' / 'static' / 'error_screenshot.png'
|
|
to_thumbnail = Image.open(error_img)
|
|
|
|
to_thumbnail.thumbnail(size)
|
|
to_thumbnail.save(to_return, 'png')
|
|
|
|
to_return.seek(0)
|
|
if for_datauri:
|
|
return base64.b64encode(to_return.getvalue()).decode()
|
|
else:
|
|
return to_return
|
|
|
|
def get_capture(self, capture_uuid: str, /) -> BytesIO:
|
|
'''Get all the files related to this capture.'''
|
|
return self._get_raw(capture_uuid)
|
|
|
|
def get_urls_rendered_page(self, capture_uuid: str, /) -> list[str]:
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
return sorted(set(ct.root_hartree.rendered_node.urls_in_rendered_page)
|
|
- set(ct.root_hartree.all_url_requests.keys()))
|
|
|
|
def compute_mmh3_shodan(self, favicon: bytes, /) -> str:
|
|
b64 = base64.encodebytes(favicon)
|
|
return str(mmh3.hash(b64))
|
|
|
|
def get_ressource(self, tree_uuid: str, /, urlnode_uuid: str, h: str | None) -> tuple[str, BytesIO, str] | None:
|
|
'''Get a specific resource from a URL node. If a hash s also given, we want an embeded resource'''
|
|
try:
|
|
url = self.get_urlnode_from_tree(tree_uuid, urlnode_uuid)
|
|
except IndexError:
|
|
# unable to find the uuid, the cache is probably in a weird state.
|
|
return None
|
|
except NoValidHarFile as e:
|
|
# something went poorly when rebuilding the tree (probably a recursive error)
|
|
self.logger.warning(e)
|
|
return None
|
|
if url.empty_response:
|
|
return None
|
|
if not h or h == url.body_hash:
|
|
# we want the body
|
|
return url.filename if url.filename else 'file.bin', BytesIO(url.body.getvalue()), url.mimetype
|
|
|
|
# We want an embedded ressource
|
|
if h not in url.resources_hashes:
|
|
return None
|
|
for mimetype, blobs in url.embedded_ressources.items():
|
|
for ressource_h, blob in blobs:
|
|
if ressource_h == h:
|
|
return 'embedded_ressource.bin', BytesIO(blob.getvalue()), mimetype
|
|
return None
|
|
|
|
def __misp_add_vt_to_URLObject(self, obj: MISPObject) -> MISPObject | None:
|
|
urls = obj.get_attributes_by_relation('url')
|
|
if not urls:
|
|
return None
|
|
url = urls[0]
|
|
self.vt.url_lookup(url.value)
|
|
report = self.vt.get_url_lookup(url.value)
|
|
if not report:
|
|
return None
|
|
vt_obj = MISPObject('virustotal-report', standalone=False)
|
|
vt_obj.add_attribute('first-submission', value=datetime.fromtimestamp(report['attributes']['first_submission_date']), disable_correlation=True)
|
|
vt_obj.add_attribute('last-submission', value=datetime.fromtimestamp(report['attributes']['last_submission_date']), disable_correlation=True)
|
|
vt_obj.add_attribute('permalink', value=f"https://www.virustotal.com/gui/url/{report['id']}/detection", disable_correlation=True)
|
|
obj.add_reference(vt_obj, 'analysed-with')
|
|
return vt_obj
|
|
|
|
def __misp_add_urlscan_to_event(self, capture_uuid: str, visibility: str) -> MISPAttribute | None:
|
|
if cache := self.capture_cache(capture_uuid):
|
|
response = self.urlscan.url_submit(cache, visibility)
|
|
if 'result' in response:
|
|
attribute = MISPAttribute()
|
|
attribute.value = response['result']
|
|
attribute.type = 'link'
|
|
return attribute
|
|
return None
|
|
|
|
def misp_export(self, capture_uuid: str, /, with_parent: bool=False) -> list[MISPEvent] | dict[str, str]:
|
|
'''Export a capture in MISP format. You can POST the return of this method
|
|
directly to a MISP instance and it will create an event.'''
|
|
cache = self.capture_cache(capture_uuid)
|
|
if not cache:
|
|
return {'error': 'UUID missing in cache, try again later.'}
|
|
|
|
# The tree is needed to generate the export. The call below makes sure it is cached
|
|
# as it may not be if the uses calls the json export without viewing the tree first,
|
|
# and it has been archived.
|
|
self.get_crawled_tree(capture_uuid)
|
|
|
|
# if the file submitted on lookyloo cannot be displayed (PDF), it will be downloaded.
|
|
# In the case, we want to have it as a FileObject in the export
|
|
filename, pseudofile = self.get_data(capture_uuid)
|
|
if filename:
|
|
event = self.misps.export(cache, self.is_public_instance, filename, pseudofile)
|
|
else:
|
|
event = self.misps.export(cache, self.is_public_instance)
|
|
screenshot: MISPAttribute = event.add_attribute('attachment', 'screenshot_landing_page.png',
|
|
data=self.get_screenshot(cache.uuid),
|
|
disable_correlation=True) # type: ignore[assignment]
|
|
# If the last object attached to tht event is a file, it is the rendered page
|
|
if event.objects and event.objects[-1].name == 'file':
|
|
event.objects[-1].add_reference(screenshot, 'rendered-as', 'Screenshot of the page')
|
|
|
|
if self.vt.available:
|
|
for e_obj in event.objects:
|
|
if e_obj.name != 'url':
|
|
continue
|
|
vt_obj = self.__misp_add_vt_to_URLObject(e_obj)
|
|
if vt_obj:
|
|
event.add_object(vt_obj)
|
|
|
|
if self.phishtank.available:
|
|
for e_obj in event.objects:
|
|
if e_obj.name != 'url':
|
|
continue
|
|
urls = e_obj.get_attributes_by_relation('url')
|
|
if not urls:
|
|
continue
|
|
pt_entry = self.phishtank.get_url_lookup(urls[0].value)
|
|
if not pt_entry or not pt_entry.get('phish_detail_url'):
|
|
continue
|
|
pt_attribute: MISPAttribute = event.add_attribute('link', value=pt_entry['phish_detail_url'], comment='Phishtank permalink') # type: ignore[assignment]
|
|
e_obj.add_reference(pt_attribute, 'known-as', 'Permalink on Phishtank')
|
|
|
|
if self.urlscan.available:
|
|
urlscan_attribute = self.__misp_add_urlscan_to_event(
|
|
capture_uuid,
|
|
visibility='unlisted' if (cache and cache.no_index) else 'public')
|
|
if urlscan_attribute:
|
|
event.add_attribute(**urlscan_attribute)
|
|
|
|
if with_parent and cache.parent:
|
|
parent = self.misp_export(cache.parent, with_parent)
|
|
if isinstance(parent, dict):
|
|
# Something bad happened
|
|
return parent
|
|
|
|
event.extends_uuid = parent[-1].uuid
|
|
parent.append(event)
|
|
return parent
|
|
|
|
return [event]
|
|
|
|
def get_misp_occurrences(self, capture_uuid: str, /, *, instance_name: str | None=None) -> tuple[dict[str, set[str]], str] | None:
|
|
if instance_name is None:
|
|
misp = self.misps.default_misp
|
|
elif self.misps.get(instance_name) is not None:
|
|
misp = self.misps[instance_name]
|
|
else:
|
|
self.logger.warning(f'MISP instance "{instance_name}" does not exists.')
|
|
return None
|
|
|
|
if not misp.available:
|
|
return None
|
|
try:
|
|
ct = self.get_crawled_tree(capture_uuid)
|
|
except LookylooException:
|
|
self.logger.warning(f'Unable to get the modules responses unless the tree ({capture_uuid}) is cached.')
|
|
return None
|
|
nodes_to_lookup = ct.root_hartree.rendered_node.get_ancestors() + [ct.root_hartree.rendered_node]
|
|
to_return: dict[str, set[str]] = defaultdict(set)
|
|
for node in nodes_to_lookup:
|
|
hits = misp.lookup(node, ct.root_hartree.get_host_node_by_uuid(node.hostnode_uuid))
|
|
for event_id, values in hits.items():
|
|
if not isinstance(values, set):
|
|
continue
|
|
to_return[event_id].update(values)
|
|
return to_return, misp.client.root_url
|
|
|
|
def get_hashes_with_context(self, tree_uuid: str, /, algorithm: str, *, urls_only: bool=False) -> dict[str, set[str]] | dict[str, list[URLNode]]:
|
|
"""Build (on demand) hashes for all the ressources of the tree, using the alorighm provided by the user.
|
|
If you just want the hashes in SHA512, use the get_hashes method, it gives you a list of hashes an they're build
|
|
with the tree. This method is computing the hashes when you query it, so it is slower."""
|
|
ct = self.get_crawled_tree(tree_uuid)
|
|
hashes = ct.root_hartree.build_all_hashes(algorithm)
|
|
if urls_only:
|
|
return {h: {node.name for node in nodes} for h, nodes in hashes.items()}
|
|
return hashes
|
|
|
|
def merge_hashlookup_tree(self, tree_uuid: str, /) -> tuple[dict[str, dict[str, Any]], int]:
|
|
if not self.hashlookup.available:
|
|
raise LookylooException('Hashlookup module not enabled.')
|
|
hashes_tree = self.get_hashes_with_context(tree_uuid, algorithm='sha1')
|
|
|
|
hashlookup_file = self._captures_index[tree_uuid].capture_dir / 'hashlookup.json'
|
|
if not hashlookup_file.exists():
|
|
ct = self.get_crawled_tree(tree_uuid)
|
|
self.hashlookup.capture_default_trigger(ct, auto_trigger=False)
|
|
|
|
if not hashlookup_file.exists():
|
|
# no hits on hashlookup
|
|
return {}, len(hashes_tree)
|
|
|
|
with hashlookup_file.open() as f:
|
|
hashlookup_entries = json.load(f)
|
|
|
|
to_return: dict[str, dict[str, Any]] = defaultdict(dict)
|
|
|
|
for sha1 in hashlookup_entries.keys():
|
|
to_return[sha1]['nodes'] = hashes_tree[sha1]
|
|
to_return[sha1]['hashlookup'] = hashlookup_entries[sha1]
|
|
return to_return, len(hashes_tree)
|
|
|
|
def get_hashes(self, tree_uuid: str, /, hostnode_uuid: str | None=None, urlnode_uuid: str | None=None) -> set[str]:
|
|
"""Return hashes (sha512) of resources.
|
|
Only tree_uuid: All the hashes
|
|
tree_uuid and hostnode_uuid: hashes of all the resources in that hostnode (including embedded ressources)
|
|
tree_uuid, hostnode_uuid, and urlnode_uuid: hash of the URL node body, and embedded resources
|
|
"""
|
|
container: CrawledTree | HostNode | URLNode
|
|
if urlnode_uuid:
|
|
container = self.get_urlnode_from_tree(tree_uuid, urlnode_uuid)
|
|
elif hostnode_uuid:
|
|
container = self.get_hostnode_from_tree(tree_uuid, hostnode_uuid)
|
|
else:
|
|
container = self.get_crawled_tree(tree_uuid)
|
|
return get_resources_hashes(container)
|
|
|
|
def get_hostnames(self, tree_uuid: str, /, hostnode_uuid: str | None=None, urlnode_uuid: str | None=None) -> set[str]:
|
|
"""Return all the unique hostnames:
|
|
* of a complete tree if no hostnode_uuid and urlnode_uuid are given
|
|
* of a HostNode if hostnode_uuid is given
|
|
* of a URLNode if urlnode_uuid is given
|
|
"""
|
|
if urlnode_uuid:
|
|
node = self.get_urlnode_from_tree(tree_uuid, urlnode_uuid)
|
|
return {node.hostname}
|
|
elif hostnode_uuid:
|
|
node = self.get_hostnode_from_tree(tree_uuid, hostnode_uuid)
|
|
return {node.name}
|
|
else:
|
|
ct = self.get_crawled_tree(tree_uuid)
|
|
return {node.name for node in ct.root_hartree.hostname_tree.traverse()}
|
|
|
|
def get_urls(self, tree_uuid: str, /, hostnode_uuid: str | None=None, urlnode_uuid: str | None=None) -> set[str]:
|
|
"""Return all the unique URLs:
|
|
* of a complete tree if no hostnode_uuid and urlnode_uuid are given
|
|
* of a HostNode if hostnode_uuid is given
|
|
* of a URLNode if urlnode_uuid is given
|
|
"""
|
|
if urlnode_uuid:
|
|
node = self.get_urlnode_from_tree(tree_uuid, urlnode_uuid)
|
|
return {node.name}
|
|
elif hostnode_uuid:
|
|
node = self.get_hostnode_from_tree(tree_uuid, hostnode_uuid)
|
|
return {urlnode.name for urlnode in node.urls}
|
|
else:
|
|
ct = self.get_crawled_tree(tree_uuid)
|
|
return {node.name for node in ct.root_hartree.url_tree.traverse()}
|
|
|
|
def get_playwright_devices(self) -> dict[str, Any]:
|
|
"""Get the preconfigured devices from Playwright"""
|
|
return get_devices()
|
|
|
|
def get_stats(self) -> dict[str, list[Any]]:
|
|
'''Gather statistics about the lookyloo instance'''
|
|
today = date.today()
|
|
calendar_week = today.isocalendar()[1]
|
|
|
|
stats_dict = {'submissions': 0, 'redirects': 0}
|
|
stats: dict[int, dict[int, dict[str, Any]]] = {}
|
|
weeks_stats: dict[int, dict[str, Any]] = {}
|
|
|
|
# Only recent captures that are not archived
|
|
for cache in self.sorted_capture_cache():
|
|
if not hasattr(cache, 'timestamp'):
|
|
continue
|
|
date_submission: datetime = cache.timestamp
|
|
|
|
if date_submission.year not in stats:
|
|
stats[date_submission.year] = {}
|
|
if date_submission.month not in stats[date_submission.year]:
|
|
stats[date_submission.year][date_submission.month] = defaultdict(dict, **stats_dict)
|
|
stats[date_submission.year][date_submission.month]['uniq_urls'] = set()
|
|
stats[date_submission.year][date_submission.month]['submissions'] += 1
|
|
stats[date_submission.year][date_submission.month]['uniq_urls'].add(cache.url)
|
|
if hasattr(cache, 'redirects') and len(cache.redirects) > 0:
|
|
stats[date_submission.year][date_submission.month]['redirects'] += len(cache.redirects)
|
|
stats[date_submission.year][date_submission.month]['uniq_urls'].update(cache.redirects)
|
|
|
|
if ((date_submission.year == today.year and calendar_week - 1 <= date_submission.isocalendar()[1] <= calendar_week)
|
|
or (calendar_week == 1 and date_submission.year == today.year - 1 and date_submission.isocalendar()[1] in [52, 53])):
|
|
if date_submission.isocalendar()[1] not in weeks_stats:
|
|
weeks_stats[date_submission.isocalendar()[1]] = defaultdict(dict, **stats_dict)
|
|
weeks_stats[date_submission.isocalendar()[1]]['uniq_urls'] = set()
|
|
weeks_stats[date_submission.isocalendar()[1]]['submissions'] += 1
|
|
weeks_stats[date_submission.isocalendar()[1]]['uniq_urls'].add(cache.url)
|
|
if hasattr(cache, 'redirects') and len(cache.redirects) > 0:
|
|
weeks_stats[date_submission.isocalendar()[1]]['redirects'] += len(cache.redirects)
|
|
weeks_stats[date_submission.isocalendar()[1]]['uniq_urls'].update(cache.redirects)
|
|
|
|
# Build limited stats based on archved captures and the indexes
|
|
for _, capture_path in self.redis.hscan_iter('lookup_dirs_archived'):
|
|
capture_ts = datetime.fromisoformat(capture_path.rsplit('/', 1)[-1])
|
|
if capture_ts.year not in stats:
|
|
stats[capture_ts.year] = {}
|
|
if capture_ts.month not in stats[capture_ts.year]:
|
|
stats[capture_ts.year][capture_ts.month] = {'submissions': 0}
|
|
stats[capture_ts.year][capture_ts.month]['submissions'] += 1
|
|
|
|
statistics: dict[str, list[Any]] = {'weeks': [], 'years': []}
|
|
for week_number in sorted(weeks_stats.keys()):
|
|
week_stat = weeks_stats[week_number]
|
|
urls = week_stat.pop('uniq_urls')
|
|
week_stat['week_number'] = week_number
|
|
week_stat['uniq_urls'] = len(urls)
|
|
week_stat['uniq_domains'] = len(uniq_domains(urls))
|
|
statistics['weeks'].append(week_stat)
|
|
|
|
for year in sorted(stats.keys()):
|
|
year_stats: dict[str, int | list[Any]] = {'year': year, 'months': [], 'yearly_submissions': 0}
|
|
for month in sorted(stats[year].keys()):
|
|
month_stats = stats[year][month]
|
|
if len(month_stats) == 1:
|
|
# archived captures, missing many values
|
|
month_stats['month_number'] = month
|
|
else:
|
|
urls = month_stats.pop('uniq_urls')
|
|
month_stats['month_number'] = month
|
|
month_stats['uniq_urls'] = len(urls)
|
|
month_stats['uniq_domains'] = len(uniq_domains(urls))
|
|
|
|
year_stats['months'].append(month_stats) # type: ignore[union-attr]
|
|
year_stats['yearly_submissions'] += month_stats['submissions']
|
|
statistics['years'].append(year_stats)
|
|
|
|
return statistics
|
|
|
|
def store_capture(self, uuid: str, is_public: bool,
|
|
os: str | None=None, browser: str | None=None,
|
|
parent: str | None=None,
|
|
downloaded_filename: str | None=None, downloaded_file: bytes | None=None,
|
|
error: str | None=None, har: dict[str, Any] | None=None,
|
|
png: bytes | None=None, html: str | None=None,
|
|
last_redirected_url: str | None=None,
|
|
cookies: list[Cookie] | list[dict[str, str]] | None=None,
|
|
capture_settings: CaptureSettings | None=None,
|
|
potential_favicons: set[bytes] | None=None
|
|
) -> None:
|
|
|
|
now = datetime.now()
|
|
dirpath = self.capture_dir / str(now.year) / f'{now.month:02}' / f'{now.day:02}' / now.isoformat()
|
|
safe_create_dir(dirpath)
|
|
|
|
if os or browser:
|
|
meta: dict[str, str] = {}
|
|
if os:
|
|
meta['os'] = os
|
|
if browser:
|
|
meta['browser'] = browser
|
|
with (dirpath / 'meta').open('w') as _meta:
|
|
json.dump(meta, _meta)
|
|
|
|
# Write UUID
|
|
with (dirpath / 'uuid').open('w') as _uuid:
|
|
_uuid.write(uuid)
|
|
|
|
# Write no_index marker (optional)
|
|
if not is_public:
|
|
(dirpath / 'no_index').touch()
|
|
|
|
# Write parent UUID (optional)
|
|
if parent:
|
|
with (dirpath / 'parent').open('w') as _parent:
|
|
_parent.write(parent)
|
|
|
|
if downloaded_filename:
|
|
with (dirpath / '0.data.filename').open('w') as _downloaded_filename:
|
|
_downloaded_filename.write(downloaded_filename)
|
|
|
|
if downloaded_file:
|
|
with (dirpath / '0.data').open('wb') as _downloaded_file:
|
|
_downloaded_file.write(downloaded_file)
|
|
|
|
if error:
|
|
with (dirpath / 'error.txt').open('w') as _error:
|
|
json.dump(error, _error)
|
|
|
|
if har:
|
|
with gzip.open(dirpath / '0.har.gz', 'wt') as f_out:
|
|
f_out.write(json.dumps(har))
|
|
|
|
if png:
|
|
with (dirpath / '0.png').open('wb') as _img:
|
|
_img.write(png)
|
|
|
|
if html:
|
|
try:
|
|
with (dirpath / '0.html').open('w') as _html:
|
|
_html.write(html)
|
|
except UnicodeEncodeError:
|
|
# NOTE: Unable to store as string, try to store as bytes instead
|
|
# Yes, it is dirty.
|
|
with (dirpath / '0.html').open('wb') as _html:
|
|
_html.write(html.encode('utf-16', 'surrogatepass'))
|
|
|
|
if last_redirected_url:
|
|
with (dirpath / '0.last_redirect.txt').open('w') as _redir:
|
|
_redir.write(last_redirected_url)
|
|
|
|
if cookies:
|
|
with (dirpath / '0.cookies.json').open('w') as _cookies:
|
|
json.dump(cookies, _cookies)
|
|
|
|
if capture_settings:
|
|
with (dirpath / 'capture_settings.json').open('w') as _cs:
|
|
json.dump(capture_settings, _cs)
|
|
|
|
if potential_favicons:
|
|
for f_id, favicon in enumerate(potential_favicons):
|
|
with (dirpath / f'{f_id}.potential_favicons.ico').open('wb') as _fw:
|
|
_fw.write(favicon)
|
|
|
|
self.redis.hset('lookup_dirs', uuid, str(dirpath))
|