Code to explore android sms corruption by comparing raw and sms tables in mmssms.db

raw_mismatches.py

#!/usr/bin/env python

import sqlite3
import pprint
import datetime
import time
# install from https://github.com/pmarti/python-messaging/tarball/master - built against 0.5.12
from messaging import sms
import pprint

# TODO: check timezones...

def adapt_datetime(ts):
    return int(time.mktime(ts.timetuple())*1000)

sqlite3.register_adapter(datetime.datetime, adapt_datetime)

class SMSExplorer(object):
    DEFAULT_DATE_RANGE = datetime.timedelta(minutes=15)
    TIMEZONE_OFFSET = datetime.timedelta(seconds=time.timezone)

    def __init__(self):
        self.db = sqlite3.connect("mmssms.db")

    @property
    def raw_addresses(self):
        """shows all the addresses in the raw table"""
        cursor = self.db.cursor()
        try:
            cursor.execute("select distinct address from raw")
            return set([values[0] for values in cursor.fetchall()])
        finally:
            cursor.close()

    def raw_entries(self, address=None):
        """returns entries from the raw table as dicts, with parsed pdu under parsed_pdu, filtering by address if given"""
        cursor = self.db.cursor()
        try:
            if address is None:
                cursor.execute("select * from raw")
            else:
                cursor.execute("select * from raw where address = ?", (address,))
            columns = [c[0] for c in cursor.description]
            entries = [dict(zip(columns, values)) for values in cursor.fetchall()]
        finally:
            cursor.close()
        for entry in entries:
            # entry["datetime"] = datetime.
            entry["parsed"] = sms.SmsDeliver(entry["pdu"])
        return entries

    def search_sms(self, address, date, date_range=DEFAULT_DATE_RANGE):
        """finds possible retrieved smses from the given address, close to the given date"""
        cursor = self.db.cursor()
        try:
            date_min, date_max = date - date_range, date + date_range
            # limit to retrieved sms from the given address in a date window
            cursor.execute("select * from sms where address = ? and date >= ? and date <= ? and type = 1", (address, date_min, date_max))
            columns = [c[0] for c in cursor.description]
            messages = [dict(zip(columns, values)) for values in cursor.fetchall()]
            for message in messages:
                message["date_parsed"] = datetime.datetime.utcfromtimestamp(message["date"]/1000.)
            return messages
        finally:
            cursor.close()

if __name__ == "__main__":
    explorer = SMSExplorer()
    for address in explorer.raw_addresses:
        entries = explorer.raw_entries(address)
        for entry in entries:
            entry_parsed = entry["parsed"]
            entry_date = entry_parsed.date
            sms_possibilities = explorer.search_sms(address, entry_date - explorer.TIMEZONE_OFFSET)
            if sms_possibilities and True not in [entry_parsed.text in s["body"] for s in sms_possibilities]:
                pprint.pprint((entry_parsed.number, entry_parsed.date, entry_parsed.text, entry_parsed.udh))
                for s in sms_possibilities:
                    pprint.pprint((s["address"], s["date_parsed"], s["body"]))
                print