Chapter 9: Advanced Search Techniques and Historical Data Recovery¶

import requests
from datetime import datetime

class WaybackMachineClient:
    """Internet Archive Wayback Machine API client"""

    API_BASE = "https://archive.org/wayback/available"
    CDX_API = "https://web.archive.org/cdx/search/cdx"

    def get_closest_snapshot(self, url, timestamp=None):
        """Get the archived version of a URL closest to a given timestamp"""
        params = {'url': url}
        if timestamp:
            params['timestamp'] = timestamp

        response = requests.get(self.API_BASE, params=params)
        if response.status_code == 200:
            data = response.json()
            if data.get('archived_snapshots', {}).get('closest'):
                return data['archived_snapshots']['closest']
        return None

    def get_all_snapshots(self, url, from_date=None, to_date=None,
                          mime_type=None, status_code=None, limit=100):
        """Get all archived snapshots for a URL"""
        params = {
            'url': url,
            'output': 'json',
            'fl': 'timestamp,original,statuscode,mimetype,length',
            'limit': limit,
        }

        if from_date:
            params['from'] = from_date  # Format: YYYYMMDD
        if to_date:
            params['to'] = to_date
        if mime_type:
            params['filter'] = f'mimetype:{mime_type}'
        if status_code:
            params['filter'] = f'statuscode:{status_code}'

        response = requests.get(self.CDX_API, params=params)
        if response.status_code == 200:
            lines = response.text.strip().split('\n')
            results = []
            for line in lines[1:]:  # Skip header
                if line:
                    parts = line.strip('[]').split(',')
                    if len(parts) >= 5:
                        results.append({
                            'timestamp': parts[0].strip('"'),
                            'original_url': parts[1].strip('"'),
                            'status_code': parts[2].strip('"'),
                            'mime_type': parts[3].strip('"'),
                            'length': parts[4].strip('"'),
                            'archive_url': f"https://web.archive.org/web/{parts[0].strip()}/{parts[1].strip('\"')}"
                        })
            return results
        return []

    def save_page_now(self, url):
        """Save a page to the Wayback Machine right now"""
        response = requests.get(
            f"https://web.archive.org/save/{url}",
            allow_redirects=False
        )

        if response.status_code == 302:
            archive_url = response.headers.get('Location', '')
            return {'status': 'saved', 'archive_url': archive_url}
        return {'status': 'failed', 'status_code': response.status_code}

    def get_domain_history(self, domain):
        """Get all archived pages for an entire domain"""
        params = {
            'url': f"{domain}/*",
            'output': 'json',
            'fl': 'timestamp,original,statuscode',
            'collapse': 'urlkey',
            'limit': 10000,
        }

        response = requests.get(self.CDX_API, params=params)
        if response.status_code == 200:
            lines = response.text.strip().split('\n')
            return [line for line in lines[1:] if line]
        return []

    def find_deleted_subpages(self, domain, keyword=None):
        """Find pages that existed on a domain but no longer do"""
        # Get all archived pages
        all_archived = self.get_domain_history(domain)

        # Check which currently 404
        deleted = []
        for entry in all_archived[:100]:  # Sample check, full check would be expensive
            try:
                parts = entry.strip('[]').split(',')
                if len(parts) >= 2:
                    url = parts[1].strip('"')
                    response = requests.get(url, timeout=5)
                    if response.status_code == 404:
                        deleted.append({'url': url, 'archived_entry': entry})
            except Exception:
                continue

        return deleted

# Usage
wayback = WaybackMachineClient()

# Find archived versions of a specific URL
snapshots = wayback.get_all_snapshots(
    'https://example.com/about',
    from_date='20180101',
    to_date='20231231'
)

# Save current state before it changes
save_result = wayback.save_page_now('https://example.com/press-release')

import requests

def search_pastebin_google(query):
    """Use Google to search pastebin content (limited effectiveness)"""
    results = []

    # Google search limited to paste sites
    sites = [
        'pastebin.com',
        'ghostbin.com',
        'paste.ee',
        'dpaste.com',
        'hastebin.com',
    ]

    for site in sites:
        google_query = f'site:{site} "{query}"'
        # Note: programmatic Google search requires API; manual execution via browser
        results.append({'site': site, 'query': google_query})

    return results

def intelligence_x_search(query, api_key, buckets=None):
    """Search Intelligence X for a query term"""

    # Step 1: Initiate search
    search_response = requests.post(
        'https://2.intelx.io/intelligent/search',
        headers={'x-key': api_key, 'Content-Type': 'application/json'},
        json={
            'term': query,
            'buckets': buckets or [],  # e.g., ['pastes', 'darkweb']
            'lookuplevel': 0,
            'maxresults': 100,
            'timeout': 5,
        }
    )

    if search_response.status_code != 200:
        return []

    search_id = search_response.json().get('id')

    # Step 2: Retrieve results
    results_response = requests.get(
        f'https://2.intelx.io/intelligent/search/result',
        headers={'x-key': api_key},
        params={'id': search_id, 'format': 1}
    )

    if results_response.status_code == 200:
        return results_response.json().get('records', [])
    return []

import subprocess
import json
from pathlib import Path

def extract_document_metadata(file_path):
    """Extract all metadata from a document using ExifTool"""
    try:
        result = subprocess.run(
            ['exiftool', '-json', '-all', str(file_path)],
            capture_output=True,
            text=True,
            timeout=30
        )

        if result.returncode == 0:
            data = json.loads(result.stdout)
            if data:
                metadata = data[0]
                # Focus on investigatively relevant fields
                relevant = {
                    'file_name': metadata.get('FileName'),
                    'file_type': metadata.get('FileType'),
                    'create_date': metadata.get('CreateDate'),
                    'modify_date': metadata.get('ModifyDate'),
                    'author': metadata.get('Author'),
                    'creator': metadata.get('Creator'),
                    'last_modified_by': metadata.get('LastModifiedBy'),
                    'company': metadata.get('Company'),
                    'title': metadata.get('Title'),
                    'subject': metadata.get('Subject'),
                    'keywords': metadata.get('Keywords'),
                    'software': metadata.get('Software') or metadata.get('Application'),
                    'revision_number': metadata.get('RevisionNumber'),
                    'total_edit_time': metadata.get('TotalEditTime'),
                    'template': metadata.get('Template'),
                    'gps_latitude': metadata.get('GPSLatitude'),
                    'gps_longitude': metadata.get('GPSLongitude'),
                }
                return {k: v for k, v in relevant.items() if v is not None}
    except Exception as e:
        return {'error': str(e)}

    return {}

def batch_metadata_extraction(directory_path, extensions=None):
    """Extract metadata from all documents in a directory"""
    extensions = extensions or ['.pdf', '.docx', '.xlsx', '.pptx', '.jpg', '.png']
    results = []

    for file_path in Path(directory_path).rglob('*'):
        if file_path.suffix.lower() in extensions:
            metadata = extract_document_metadata(file_path)
            if metadata and 'error' not in metadata:
                results.append({
                    'file': str(file_path),
                    'metadata': metadata
                })

    return results

# Analyze for patterns across multiple documents
def analyze_document_set(metadata_list):
    """Find patterns across a set of documents (e.g., leaked corporate files)"""
    from collections import Counter

    authors = Counter()
    companies = Counter()
    software = Counter()
    dates = []

    for item in metadata_list:
        m = item['metadata']
        if m.get('author'):
            authors[m['author']] += 1
        if m.get('company'):
            companies[m['company']] += 1
        if m.get('software'):
            software[m['software']] += 1
        if m.get('create_date'):
            dates.append(m['create_date'])

    return {
        'top_authors': authors.most_common(10),
        'companies': companies.most_common(10),
        'software_used': software.most_common(10),
        'date_range': (min(dates), max(dates)) if dates else None,
    }

import json
from datetime import datetime
import itertools

class SearchWorkflow:
    """Systematic search workflow manager"""

    def __init__(self, investigation_name):
        self.investigation = investigation_name
        self.queries_run = []
        self.results = []
        self.coverage_log = {}

    def generate_query_matrix(self, names, companies, keywords, operators=None):
        """Generate systematic query combinations"""
        queries = []

        # Base name queries
        for name in names:
            queries.append(f'"{name}"')

        # Name + company combinations
        for name, company in itertools.product(names, companies):
            queries.append(f'"{name}" "{company}"')

        # Name + keyword combinations
        for name, keyword in itertools.product(names, keywords):
            queries.append(f'"{name}" "{keyword}"')

        # Site-specific queries
        sites = ['linkedin.com', 'twitter.com', 'facebook.com', 'instagram.com']
        for name, site in itertools.product(names, sites):
            queries.append(f'site:{site} "{name}"')

        # File type queries for documents
        for name in names:
            for filetype in ['pdf', 'docx', 'xlsx']:
                queries.append(f'"{name}" filetype:{filetype}')

        return queries

    def log_search(self, query, engine, results_count, relevant_count, notes=""):
        """Log a search execution"""
        entry = {
            'timestamp': datetime.now().isoformat(),
            'query': query,
            'engine': engine,
            'results_count': results_count,
            'relevant_count': relevant_count,
            'notes': notes
        }
        self.queries_run.append(entry)
        return entry

    def identify_coverage_gaps(self, subjects, time_periods):
        """Identify which subject/period combinations haven't been searched"""
        gaps = []
        for subject in subjects:
            for period in time_periods:
                key = f"{subject}_{period}"
                if key not in self.coverage_log:
                    gaps.append({'subject': subject, 'period': period})
        return gaps

    def export_search_log(self, output_path):
        """Export search log for documentation"""
        with open(output_path, 'w') as f:
            json.dump({
                'investigation': self.investigation,
                'total_queries': len(self.queries_run),
                'queries': self.queries_run,
            }, f, indent=2)