hibp.ipynb

{
 "metadata": {
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.9.0-final"
  },
  "orig_nbformat": 2,
  "kernelspec": {
   "name": "Python 3.9.0 64-bit",
   "display_name": "Python 3.9.0 64-bit",
   "metadata": {
    "interpreter": {
     "hash": "7812ea015bdcee6f23a998adcdd2ef97c151c0c241b7b7070987d9313e41299d"
    }
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2,
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Have Py Been Pawned?\n",
    "\n",
    "## Overview: \n",
    "[HaveIBeenPawned](https://haveibeenpwned.com/) is a well-known website which aggregegates email breach data. It allows people to search their email address via the website and they also have an API.\n",
    "\n",
    "This script and jupyter notebook allow you to run your own list of emails against the API to find out if they appear in any breaches that HIBP have loaded. \n",
    "\n",
    "## What this repo contains\n",
    "A jupyer notebook and an equivalent script\n",
    "\n",
    "## Why did I build it?\n",
    "1. It's handy to be able to find out whether a list of email addresses appear in any data breaches\n",
    "2. HaveIBeenPawned doesn't support bulk email address lookup as of February 2021\n",
    "3. It's easy to extend this and analyse the addresses further with pandas \n",
    " \n",
    "There are a couple of scripts floating around but I didn't like the way they were written, or they used superseded API or Python versions. \n",
    "\n",
    "Thought someone else might find this useful, so figured I'd share it.\n",
    "\n",
    "## Whats it do?: \n",
    "This tool takes a csv of email addresses and hits the HaveIBeenPawned API to find out whether those email addresses appear in any breaches\n",
    "\n",
    "## Requirements:\n",
    "- Python 3.9 (probably works on 3xx but I haven't tested it)\n",
    "- Some standard python libs\n",
    "- An HIBP API Key https://haveibeenpwned.com/API/Key \n",
    "  - *(Don't forget to read the docs https://haveibeenpwned.com/API/v3#Authorisation)*\n",
    "- A csv file containing email addresses in the first column \n",
    "\n",
    "## How to use it \n",
    "\n",
    "First, set up a virtual environment\n",
    "\n",
    "```console\n",
    "foo@bar:~$ python3 -m venv env\n",
    "```\n",
    "\n",
    "Then activate it\n",
    "```console\n",
    "foo@bar:~$ source env/bin/activate\n",
    "```\n",
    "\n",
    "### For the Jupyter Notebook\n",
    "In the Second Cell: \n",
    "\n",
    "1. Update the `fileLocation` variable to the location of your email source file\n",
    "\n",
    "2. Update the `apikey` and `user-agent` variables to your HIBP API Key and a relevant user-agent per HIBP docs https://haveibeenpwned.com/API/v3#Authorisation \n",
    "\n",
    "Run all cells"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import csv\n",
    "import requests\n",
    "import time "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Settings\n",
    "fileLocation = 'test.csv'\n",
    "resultFileLocation = 'hibp_outcome.csv'\n",
    "includeUnverifiedBreaches = 'true'\n",
    "apikey = ''\n",
    "userAgent = ''\n",
    "printResults = True\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Turn csv of emails into a deduped dict to work with\n",
    "def getEmailAddressesFromCSV(csvName):\n",
    "    emailAddresses = []\n",
    "    \n",
    "    # read csv file as a list of lists\n",
    "    with open(csvName, 'r') as read_obj:\n",
    "        # pass the file object to reader() to get the reader object\n",
    "        csv_reader = csv.reader(read_obj)\n",
    "        # Pass reader object to list() to get a list of lists\n",
    "        # This is handy in case you want to use a csv with more columns\n",
    "        list_of_rows = list(csv_reader)\n",
    "\n",
    "    # Quick check for duplicate addresses \n",
    "    for r in list_of_rows:\n",
    "        if r[0] not in emailAddresses:\n",
    "            emailAddresses.append(r[0])\n",
    "        else: \n",
    "            pass \n",
    "        \n",
    "    return emailAddresses\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "wait = 1.3\n",
    "reqUrl = 'https://haveibeenpwned.com/api/v3/breachedaccount/'\n",
    "headers = {'hibp-api-key': apikey, 'user-agent': userAgent}\n",
    "\n",
    "def makeRequest(email):\n",
    "    # sleep = wait \n",
    "    # initialize the request\n",
    "    check = requests.get(reqUrl+ email + '?includeUnverified=' + includeUnverifiedBreaches,\n",
    "                 headers=headers)\n",
    "    return check\n",
    "\n",
    "def checkHIBP(email):\n",
    "    # List to hold outcome of check\n",
    "    hibpResult = [email]\n",
    "    # takes wait time setting \n",
    "    sleep = wait \n",
    "    # initialize the request\n",
    "    check = makeRequest(email)\n",
    "    # retry if rate limited - shouldn't happen for anything over 1.3 seconds\n",
    "    while str(check.status_code) == '429':\n",
    "        print('Too fast, slow down! (Rate limited... will retry after ' + check.headers['Retry-After'] + ' seconds)'  ) \n",
    "        sleep = float(check.headers['Retry-After']) \n",
    "        time.sleep(sleep) \n",
    "        check = makeRequest(email)\n",
    "\n",
    "    # for some reason the HIBP API wraps the response body in b''\n",
    "    # this function strips that out\n",
    "    cleanUp = lambda check: str(check).strip('b\\'')\n",
    "    \n",
    "    # 404 indicates no breach found \n",
    "    if str(check.status_code) == '404':\n",
    "        outcome = 'no breach found'\n",
    "        if printResults: print(email + \": \" + outcome)\n",
    "        # appends emails pwnage status\n",
    "        hibpResult.append(outcome)\n",
    "        # appends 0 to indiate FALSE \n",
    "        hibpResult.append(0)\n",
    "        # appends 0 as null \n",
    "        hibpResult.append(0)\n",
    "        time.sleep(sleep) \n",
    "\n",
    "    # 200 indicates a breach\n",
    "    elif str(check.status_code) == '200': \n",
    "        outcome = 'breach found'\n",
    "        if printResults: print(email + \": \" + outcome)\n",
    "        # appened the emails pwnage status\n",
    "        hibpResult.append(outcome)\n",
    "        # appends 1 to indicate TRUE\n",
    "        hibpResult.append(1)\n",
    "        # appends body of HIPB response \n",
    "        hibpResult.append(cleanUp(check.content))\n",
    "        time.sleep(sleep) \n",
    "\n",
    "    # something else happened... you'll need to debug this yourself\n",
    "    else: \n",
    "        print(str(check.status_code))\n",
    "        print(\"something went wrong... \")\n",
    "        time.sleep(sleep) \n",
    "\n",
    "    return hibpResult\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Writes the output of the checks to a new csv\n",
    "def writeFile(out):\n",
    "    with open(resultFileLocation, 'w', newline='') as file:\n",
    "        writer = csv.writer(file)\n",
    "        writer.writerow(['email_address', 'search_outcome', 'breach_status', 'breach_detail'])\n",
    "        writer.writerows(out)\n",
    "\n",
    "# Process the various functions\n",
    "def runChecks():\n",
    "    emails = getEmailAddressesFromCSV(fileLocation)\n",
    "    print('[!] Loaded ' + '\\'' + fileLocation + '\\'')\n",
    "\n",
    "    print('[!] Beginning processing ' + '\\'' + fileLocation + '\\'' + ' against hibp api')\n",
    "    print('')\n",
    "    outcome = []\n",
    "    for e in emails:\n",
    "        # Checks for a header row in my data called 'email_address' \n",
    "        # Not necessary if you follow the instructions, but leaving it just in case ;-)\n",
    "        if e != 'email_address':\n",
    "            outcome.append(checkHIBP(e))\n",
    "    \n",
    "    print('')\n",
    "    print('[!] Finished Processing ' + '\\'' + fileLocation + '\\'' + ' against hibp api')\n",
    "    print('[!] Writing results to ' + '\\'' + resultFileLocation + '\\'' +  ' ... ')\n",
    "\n",
    "    writeFile(outcome)\n",
    "    print('[!] DONE ')\n",
    "    \n",
    "    # If you would like to use pandas to continue to analyse the file in a notebook\n",
    "    # uncomment \"return Outcome\" and initialise the dataframe with it \n",
    "    # return outcome\n",
    "\n",
    "runChecks()"
   ]
  }
 ]
}