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)