import os
import shutil
import sqlite3
import time
import socket
import tempfile
from datetime import datetime, timedelta
from elasticsearch import Elasticsearch, helpers
import schedule
from hashlib import md5

# Configuration
ELASTICSEARCH_HOST = '***'
INDEX_NAME = 'browser_history'
HOSTNAME = socket.gethostname()

ELASTIC_USER = os.getenv('ELASTIC_USER', '***')
ELASTIC_PASSWORD = os.getenv('ELASTIC_PASSWORD', '***')

# Potential Firefox and Brave history paths (check OS and possible profile variations)
FIREFOX_PATHS = [
    os.path.expanduser('~/.mozilla/firefox'),  # Linux
    os.path.expanduser('~/AppData/Roaming/Mozilla/Firefox/Profiles'),  # Windows
]

BRAVE_PATHS = [
    os.path.expanduser('~/.config/BraveSoftware/Brave-Browser/Default'),  # Linux
    os.path.expanduser('~/AppData/Local/BraveSoftware/Brave-Browser/User Data/Default'),  # Windows
]

# Transition type mappings
FIREFOX_TRANSITION_TYPES = {
    1: "link",
    2: "typed",
    3: "bookmark",
    4: "auto_subframe",
    5: "permanent_redirect",
    6: "temporary_redirect",
    7: "reload",
    8: "manual_subframe"
}

CHROME_TRANSITION_TYPES = {
    0: "link",
    1: "typed",
    2: "bookmark",
    3: "auto_subframe",
    4: "manual_subframe",
    5: "generated",
    6: "auto_toplevel",
    7: "form_submit",
    8: "reload",
    9: "keyword",
    10: "keyword_generated"
}

# Elasticsearch client
es = Elasticsearch([ELASTICSEARCH_HOST], verify_certs=False, http_auth=(ELASTIC_USER, ELASTIC_PASSWORD))

def find_history_file(possible_paths, filename):
    """Finds the first valid path containing the specified filename."""
    for path in possible_paths:
        for root, dirs, files in os.walk(path):
            if filename in files:
                return os.path.join(root, filename)
    return None

def copy_file(src):
    """Copies a file to a temporary location and returns the temporary path."""
    temp_file = tempfile.NamedTemporaryFile(delete=False)
    shutil.copy2(src, temp_file.name)
    temp_file.close()
    return temp_file.name

def check_table_exists(conn, table_name):
    cursor = conn.cursor()
    cursor.execute("SELECT name FROM sqlite_master WHERE type='table' AND name=?", (table_name,))
    return cursor.fetchone() is not None

def extract_firefox_history():
    db_path = find_history_file(FIREFOX_PATHS, 'places.sqlite')
    if not db_path:
        print('Firefox profile path does not exist')
        return []

    temp_db_path = copy_file(db_path)

    conn = sqlite3.connect(temp_db_path)
    cursor = conn.cursor()

    query = '''
    SELECT moz_places.url, moz_places.title, moz_places.visit_count, moz_historyvisits.visit_date, moz_historyvisits.from_visit, moz_historyvisits.visit_type, moz_historyvisits.id
    FROM moz_places JOIN moz_historyvisits ON moz_places.id = moz_historyvisits.place_id
    ORDER BY moz_historyvisits.visit_date DESC
    '''
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    os.remove(temp_db_path)  # remove temporary DB

    history = []
    for url, title, visit_count, visit_date, from_visit, visit_type, visit_id in results:
        visit_type = FIREFOX_TRANSITION_TYPES.get(visit_type, "Unknown")
        visit_date = datetime(1970, 1, 1) + timedelta(microseconds=visit_date)
        history.append({
            "url": url,
            "title": title,
            "visit_count": visit_count,
            "visit_time": visit_date.isoformat(),
            "from_visit": 'f_' + str(from_visit),
            "visit_type": visit_type,
            "visit_id": 'f_' + str(visit_id)
        })

    return history

def extract_brave_history():
    db_path = find_history_file(BRAVE_PATHS, 'History')
    if not db_path:
        print('Brave profile path does not exist')
        return []

    temp_db_path = copy_file(db_path)

    conn = sqlite3.connect(temp_db_path)
    cursor = conn.cursor()

    query = '''
    SELECT urls.url, urls.title, urls.visit_count, visits.visit_time, visits.from_visit, visits.transition, visits.id
    FROM urls JOIN visits ON urls.id = visits.url
    ORDER BY visits.visit_time DESC
    '''
    cursor.execute(query)
    results = cursor.fetchall()
    conn.close()
    os.remove(temp_db_path)  # remove temporary DB

    history = []
    for url, title, visit_count, visit_time, from_visit, transition, visit_id in results:
        visit_type = CHROME_TRANSITION_TYPES.get(transition & 0xFF, "Unknown")
        visit_time = datetime(1601, 1, 1) + timedelta(microseconds=visit_time)
        history.append({
            "url": url,
            "title": title,
            "visit_count": visit_count,
            "visit_time": visit_time.isoformat(),
            "from_visit": 'b_' + str(from_visit),
            "visit_type": visit_type,
            "visit_id": 'b_' + str(visit_id)
        })

    return history

def send_to_elasticsearch_bulk(history, browser):
    actions = []
    for entry in history:
        print(entry)
        # Generate a unique ID using URL, browser, visit_time, and hostname
        unique_string = f"{entry['url']}_{entry['visit_time']}"
        unique_id = md5(unique_string.encode()).hexdigest()

        # Define the document with a unique ID
        actions.append({
            "_index": INDEX_NAME,
            "_id": unique_id,  # Set the unique document ID
            "_source": {
                'url': entry['url'],
                'title': entry['title'],
                'visit_count': entry['visit_count'],
                'visit_time': entry['visit_time'],
                'from_visit': entry['from_visit'],
                'visit_type': entry['visit_type'],
                'visit_id': entry['visit_id'],
                'browser': browser,
                'host.name': HOSTNAME,
                'tags': ['browser_history', browser]
            }
        })
    helpers.bulk(es, actions)

def main():
    firefox_history = extract_firefox_history()
    brave_history = extract_brave_history()

    send_to_elasticsearch_bulk(firefox_history, 'firefox')
    send_to_elasticsearch_bulk(brave_history, 'brave')

# Schedule the script to run at regular intervals
schedule.every().hour.do(main)

if __name__ == '__main__':
    main()
    while True:
        schedule.run_pending()
        time.sleep(1)
Edit

Pub: 11 Jan 2025 19:47 UTC

Views: 136