complete.ipynb

   1 {
   2  "cells": [
   3   {
   4    "cell_type": "code",
   5    "execution_count": 1,
   6    "metadata": {
   7     "collapsed": true
   8    },
   9    "outputs": [],
  10    "source": [
  11     "import configparser\n",
  12     "import imaplib\n",
  13     "import email\n",
  14     "import io\n",
  15     "import zipfile\n",
  16     "import xml.etree.ElementTree\n",
  17     "import psycopg2\n",
  18     "import re\n",
  19     "import datetime"
  20    ]
  21   },
  22   {
  23    "cell_type": "code",
  24    "execution_count": 2,
  25    "metadata": {
  26     "collapsed": true
  27    },
  28    "outputs": [],
  29    "source": [
  30     "def fetch_msg(num):\n",
  31     "    return mailbox.uid('FETCH', num, '(RFC822)')[1][0][1]"
  32    ]
  33   },
  34   {
  35    "cell_type": "code",
  36    "execution_count": 3,
  37    "metadata": {
  38     "collapsed": true
  39    },
  40    "outputs": [],
  41    "source": [
  42     "def xml_of_part(part):\n",
  43     "    with zipfile.ZipFile(io.BytesIO(part.get_payload(decode=True))) as zf:\n",
  44     "            fn = zf.infolist()[0].filename\n",
  45     "            contents = zf.read(fn).decode('utf-8')\n",
  46     "            return xml.etree.ElementTree.fromstring(contents)"
  47    ]
  48   },
  49   {
  50    "cell_type": "code",
  51    "execution_count": 4,
  52    "metadata": {
  53     "collapsed": true
  54    },
  55    "outputs": [],
  56    "source": [
  57     "def xml_of(message):\n",
  58     "    reports = []\n",
  59     "    if message.is_multipart():\n",
  60     "        for p in message.get_payload():\n",
  61     "            if 'zip' in p.get_content_type():\n",
  62     "                reports += [xml_of_part(p)]\n",
  63     "    else:\n",
  64     "        reports = [xml_of_part(message)]\n",
  65     "    return reports"
  66    ]
  67   },
  68   {
  69    "cell_type": "code",
  70    "execution_count": 5,
  71    "metadata": {
  72     "collapsed": true
  73    },
  74    "outputs": [],
  75    "source": [
  76     "def extract_report(msg):\n",
  77     "    pmsg = email.message_from_bytes(msg)\n",
  78     "    return xml_of(pmsg)"
  79    ]
  80   },
  81   {
  82    "cell_type": "code",
  83    "execution_count": 6,
  84    "metadata": {
  85     "collapsed": true
  86    },
  87    "outputs": [],
  88    "source": [
  89     "def maybe_strip(text):\n",
  90     "    if text:\n",
  91     "        return text.strip()\n",
  92     "    else:\n",
  93     "        return ''"
  94    ]
  95   },
  96   {
  97    "cell_type": "code",
  98    "execution_count": 7,
  99    "metadata": {
 100     "collapsed": true
 101    },
 102    "outputs": [],
 103    "source": [
 104     "field_maps = {'./policy_published/adkim': {'pg_field_name': 'policy_published_adkim',\n",
 105     "  'pg_table': 'reports',\n",
 106     "  'pg_type': 'varchar'},\n",
 107     " './policy_published/aspf': {'pg_field_name': 'policy_published_aspf',\n",
 108     "  'pg_table': 'reports',\n",
 109     "  'pg_type': 'varchar'},\n",
 110     " './policy_published/domain': {'pg_field_name': 'policy_published_domain',\n",
 111     "  'pg_table': 'reports',\n",
 112     "  'pg_type': 'varchar'},\n",
 113     " './policy_published/p': {'pg_field_name': 'policy_published_p',\n",
 114     "  'pg_table': 'reports',\n",
 115     "  'pg_type': 'varchar'},\n",
 116     " './policy_published/pct': {'pg_field_name': 'policy_published_pct',\n",
 117     "  'pg_table': 'reports',\n",
 118     "  'pg_type': 'int'},\n",
 119     " './record[{}]/auth_results/dkim/domain': {'pg_field_name': 'auth_results_dkim_domain',\n",
 120     "  'pg_table': 'report_items',\n",
 121     "  'pg_type': 'varchar'},\n",
 122     " './record[{}]/auth_results/dkim/result': {'pg_field_name': 'auth_results_dkim_result',\n",
 123     "  'pg_table': 'report_items',\n",
 124     "  'pg_type': 'varchar'},\n",
 125     " './record[{}]/auth_results/spf/domain': {'pg_field_name': 'auth_results_spf_domain',\n",
 126     "  'pg_table': 'report_items',\n",
 127     "  'pg_type': 'varchar'},\n",
 128     " './record[{}]/auth_results/spf/result': {'pg_field_name': 'auth_results_spf_result',\n",
 129     "  'pg_table': 'report_items',\n",
 130     "  'pg_type': 'varchar'},\n",
 131     " './record[{}]/identifiers/header_from': {'pg_field_name': 'identifiers_header_from',\n",
 132     "  'pg_table': 'report_items',\n",
 133     "  'pg_type': 'varchar'},\n",
 134     " './record[{}]/row/count': {'pg_field_name': 'count',\n",
 135     "  'pg_table': 'report_items',\n",
 136     "  'pg_type': 'int'},\n",
 137     " './record[{}]/row/policy_evaluated/disposition': {'pg_field_name': 'policy_evaluated_disposition',\n",
 138     "  'pg_table': 'report_items',\n",
 139     "  'pg_type': 'varchar'},\n",
 140     " './record[{}]/row/policy_evaluated/dkim': {'pg_field_name': 'policy_evaluated_dkim',\n",
 141     "  'pg_table': 'report_items',\n",
 142     "  'pg_type': 'varchar'},\n",
 143     " './record[{}]/row/policy_evaluated/spf': {'pg_field_name': 'policy_evaluated_spf',\n",
 144     "  'pg_table': 'report_items',\n",
 145     "  'pg_type': 'varchar'},\n",
 146     " './record[{}]/row/source_ip': {'pg_field_name': 'source_ip',\n",
 147     "  'pg_table': 'report_items',\n",
 148     "  'pg_type': 'inet'},\n",
 149     " './report_metadata/date_range/begin': {'pg_field_name': 'report_metadata_date_range_begin',\n",
 150     "  'pg_table': 'reports',\n",
 151     "  'pg_type': 'timestamp'},\n",
 152     " './report_metadata/date_range/end': {'pg_field_name': 'report_metadata_date_range_end',\n",
 153     "  'pg_table': 'reports',\n",
 154     "  'pg_type': 'timestamp'},\n",
 155     " './report_metadata/email': {'pg_field_name': 'report_metadata_email',\n",
 156     "  'pg_table': 'reports',\n",
 157     "  'pg_type': 'varchar'},\n",
 158     " './report_metadata/org_name': {'pg_field_name': 'report_metadata_org_name',\n",
 159     "  'pg_table': 'reports',\n",
 160     "  'pg_type': 'varchar'},\n",
 161     " './report_metadata/report_id': {'pg_field_name': 'report_metadata_report_id',\n",
 162     "  'pg_table': 'reports',\n",
 163     "  'pg_type': 'varchar'}}"
 164    ]
 165   },
 166   {
 167    "cell_type": "code",
 168    "execution_count": 8,
 169    "metadata": {
 170     "collapsed": true
 171    },
 172    "outputs": [],
 173    "source": [
 174     "def build_insert_command(table_name, report, preamble_values=None, i=None):\n",
 175     "        field_names = []\n",
 176     "        if preamble_values:\n",
 177     "            values = preamble_values.copy()\n",
 178     "        else:\n",
 179     "            values = {}\n",
 180     "        for f in [f for f in field_maps if field_maps[f]['pg_table'] == table_name]:\n",
 181     "            # print(f)\n",
 182     "            if i:\n",
 183     "                fp = f.format(i)\n",
 184     "            else:\n",
 185     "                fp = f\n",
 186     "            field_names += [field_maps[f]['pg_field_name']]\n",
 187     "            if field_maps[f]['pg_type'] == 'int':\n",
 188     "                values[field_maps[f]['pg_field_name']] = int(report.find(fp).text)\n",
 189     "            elif field_maps[f]['pg_type'] == 'timestamp':\n",
 190     "                values[field_maps[f]['pg_field_name']] = datetime.datetime.utcfromtimestamp(int(report.find(fp).text))\n",
 191     "            elif field_maps[f]['pg_type'] == 'inet':\n",
 192     "                values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(fp).text)\n",
 193     "            else:\n",
 194     "                values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(fp).text)\n",
 195     "        insert_string = 'insert into {} ('.format(table_name)\n",
 196     "        if preamble_values:\n",
 197     "            insert_string += ', '.join(sorted(preamble_values.keys())) + ', '\n",
 198     "        insert_string += ', '.join(field_names) + ') '\n",
 199     "        insert_string += 'values ('\n",
 200     "        if preamble_values:\n",
 201     "            insert_string += ', '.join('%({})s'.format(fn) for fn in sorted(preamble_values.keys())) + ', '\n",
 202     "        insert_string += ', '.join('%({})s'.format(f) for f in field_names) + ');'\n",
 203     "        return insert_string, values"
 204    ]
 205   },
 206   {
 207    "cell_type": "code",
 208    "execution_count": 9,
 209    "metadata": {
 210     "collapsed": true
 211    },
 212    "outputs": [],
 213    "source": [
 214     "def write_report(connection, cursor, report):\n",
 215     "    insert_string, values = build_insert_command('reports', report)\n",
 216     "    # print(insert_string, values)\n",
 217     "    cursor.execute(insert_string, values)\n",
 218     "    \n",
 219     "    for i in range(1, len(report.findall('./record'))+1):\n",
 220     "        field_names = []\n",
 221     "        cursor.execute('select id, report_metadata_report_id from reports where report_metadata_report_id = %s;', \n",
 222     "            [report.find('./report_metadata/report_id').text])\n",
 223     "        results = cursor.fetchall()\n",
 224     "        if len(results) != 1:\n",
 225     "            raise RuntimeError('Could not find report record for report item')\n",
 226     "        else:\n",
 227     "            report_id = results[0][0]\n",
 228     "        insert_string, values = build_insert_command('report_items', report, i=i,\n",
 229     "                                                     preamble_values={'report_id': report_id})\n",
 230     "        # print(insert_string, values)\n",
 231     "        cursor.execute(insert_string, values)\n",
 232     "    connection.commit()"
 233    ]
 234   },
 235   {
 236    "cell_type": "code",
 237    "execution_count": 10,
 238    "metadata": {
 239     "collapsed": true
 240    },
 241    "outputs": [],
 242    "source": [
 243     "def write_report_old(connection, cursor, report):\n",
 244     "    \n",
 245     "    field_names = []\n",
 246     "    values = {}\n",
 247     "    for f in [f for f in field_maps if field_maps[f]['pg_table'] == 'reports']:\n",
 248     "        field_names += [field_maps[f]['pg_field_name']]\n",
 249     "        if field_maps[f]['pg_type'] == 'int':\n",
 250     "            values[field_maps[f]['pg_field_name']] = int(report.find(f).text)\n",
 251     "        elif field_maps[f]['pg_type'] == 'timestamp':\n",
 252     "            values[field_maps[f]['pg_field_name']] = datetime.datetime.utcfromtimestamp(int(report.find(f).text))\n",
 253     "        elif field_maps[f]['pg_type'] == 'inet':\n",
 254     "            values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(f).text)\n",
 255     "        else:\n",
 256     "            values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(f).text)\n",
 257     "    insert_string = 'insert into reports (' + ', '.join(field_names) + ') '\n",
 258     "    insert_string += 'values (' + ', '.join('%({})s'.format(f) for f in field_names) + ');'\n",
 259     "    # print(insert_string, values)\n",
 260     "    cursor.execute(insert_string, values)\n",
 261     "    \n",
 262     "    for i in range(1, len(report.findall('./record'))+1):\n",
 263     "        field_names = []\n",
 264     "        cursor.execute('select id, report_metadata_report_id from reports where report_metadata_report_id = %s;', \n",
 265     "            [report.find('./report_metadata/report_id').text])\n",
 266     "        results = cursor.fetchall()\n",
 267     "        if len(results) != 1:\n",
 268     "            raise\n",
 269     "        else:\n",
 270     "            report_id = results[0][0]\n",
 271     "        values = {'report_id': report_id}\n",
 272     "        for f in [f for f in field_maps if field_maps[f]['pg_table'] == 'report_items']:\n",
 273     "            field_names += [field_maps[f]['pg_field_name']]\n",
 274     "            if field_maps[f]['pg_type'] == 'int':\n",
 275     "                values[field_maps[f]['pg_field_name']] = int(report.find(f.format(i)).text)\n",
 276     "            elif field_maps[f]['pg_type'] == 'timestamp':\n",
 277     "                values[field_maps[f]['pg_field_name']] = datetime.datetime.utcfromtimestamp(int(report.find(f.format(i)).text))\n",
 278     "            elif field_maps[f]['pg_type'] == 'inet':\n",
 279     "                values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(f.format(i)).text)\n",
 280     "            else:\n",
 281     "                values[field_maps[f]['pg_field_name']] = maybe_strip(report.find(f.format(i)).text)\n",
 282     "        insert_string = 'insert into report_items (report_id, ' + ', '.join(field_names) + ') '\n",
 283     "        insert_string += 'values (%(report_id)s, ' + ', '.join('%({})s'.format(f) for f in field_names) + ');'\n",
 284     "        # print(insert_string, values)\n",
 285     "        cursor.execute(insert_string, values)\n",
 286     "    connection.commit()"
 287    ]
 288   },
 289   {
 290    "cell_type": "code",
 291    "execution_count": 11,
 292    "metadata": {
 293     "collapsed": false
 294    },
 295    "outputs": [
 296     {
 297      "data": {
 298       "text/plain": [
 299        "['dmarc.ini']"
 300       ]
 301      },
 302      "execution_count": 11,
 303      "metadata": {},
 304      "output_type": "execute_result"
 305     }
 306    ],
 307    "source": [
 308     "config = configparser.ConfigParser()\n",
 309     "config.read('dmarc.ini')"
 310    ]
 311   },
 312   {
 313    "cell_type": "code",
 314    "execution_count": 12,
 315    "metadata": {
 316     "collapsed": false
 317    },
 318    "outputs": [],
 319    "source": [
 320     "conn = psycopg2.connect(host=config['database']['server'],\n",
 321     "                        database=config['database']['database'], \n",
 322     "                        user=config['database']['username'], \n",
 323     "                        password=config['database']['password'])"
 324    ]
 325   },
 326   {
 327    "cell_type": "code",
 328    "execution_count": 13,
 329    "metadata": {
 330     "collapsed": false
 331    },
 332    "outputs": [
 333     {
 334      "data": {
 335       "text/plain": [
 336        "datetime.datetime(2016, 3, 30, 23, 59, 59)"
 337       ]
 338      },
 339      "execution_count": 13,
 340      "metadata": {},
 341      "output_type": "execute_result"
 342     }
 343    ],
 344    "source": [
 345     "with conn:\n",
 346     "    with conn.cursor() as cur:\n",
 347     "        cur.execute('select max(report_metadata_date_range_end) from reports')\n",
 348     "        results = cur.fetchall()\n",
 349     "most_recent_date = results[0][0]\n",
 350     "most_recent_date"
 351    ]
 352   },
 353   {
 354    "cell_type": "code",
 355    "execution_count": 14,
 356    "metadata": {
 357     "collapsed": false
 358    },
 359    "outputs": [
 360     {
 361      "data": {
 362       "text/plain": [
 363        "('OK', [b'179'])"
 364       ]
 365      },
 366      "execution_count": 14,
 367      "metadata": {},
 368      "output_type": "execute_result"
 369     }
 370    ],
 371    "source": [
 372     "mailbox = imaplib.IMAP4(host=config['imap']['server'], \n",
 373     "                      port=config['imap']['port'])\n",
 374     "mailbox.starttls()\n",
 375     "mailbox.login(config['imap']['username'], config['imap']['password'])\n",
 376     "mailbox.select('INBOX', readonly=True)"
 377    ]
 378   },
 379   {
 380    "cell_type": "code",
 381    "execution_count": 15,
 382    "metadata": {
 383     "collapsed": false
 384    },
 385    "outputs": [
 386     {
 387      "data": {
 388       "text/plain": [
 389        "('SINCE 28-Mar-2016', 'OK', [b'172 173 174 175 176 177 178 179 180'])"
 390       ]
 391      },
 392      "execution_count": 15,
 393      "metadata": {},
 394      "output_type": "execute_result"
 395     }
 396    ],
 397    "source": [
 398     "if most_recent_date:\n",
 399     "    mails_from = \"SINCE \" + (most_recent_date - datetime.timedelta(days=2)).strftime(\"%d-%b-%Y\")\n",
 400     "else:\n",
 401     "    mails_from = \"ALL\"\n",
 402     "resp, nums = mailbox.uid('SEARCH', None, mails_from)\n",
 403     "mails_from, resp, nums"
 404    ]
 405   },
 406   {
 407    "cell_type": "code",
 408    "execution_count": 16,
 409    "metadata": {
 410     "collapsed": false
 411    },
 412    "outputs": [
 413     {
 414      "data": {
 415       "text/plain": [
 416        "['1459129809.695034',\n",
 417        " '16143280651570354241',\n",
 418        " '8c177254c3cb41869dc3afab59f74c76@hotmail.com',\n",
 419        " '15410706527896810898',\n",
 420        " '1459216304.582931',\n",
 421        " '15497495941279624940',\n",
 422        " '1459302353.261157',\n",
 423        " '7773a696f4a54f1e8c01f4644fbb94ee@hotmail.com',\n",
 424        " '15185964531645951164']"
 425       ]
 426      },
 427      "execution_count": 16,
 428      "metadata": {},
 429      "output_type": "execute_result"
 430     }
 431    ],
 432    "source": [
 433     "dmarc_reports = [report for report_set in [extract_report(fetch_msg(n)) for n in nums[0].split()]\n",
 434     "                for report in report_set]\n",
 435     "[r.find('./report_metadata/report_id').text for r in dmarc_reports]"
 436    ]
 437   },
 438   {
 439    "cell_type": "code",
 440    "execution_count": 17,
 441    "metadata": {
 442     "collapsed": false
 443    },
 444    "outputs": [
 445     {
 446      "data": {
 447       "text/plain": [
 448        "('BYE', [b'Logging out'])"
 449       ]
 450      },
 451      "execution_count": 17,
 452      "metadata": {},
 453      "output_type": "execute_result"
 454     }
 455    ],
 456    "source": [
 457     "mailbox.close()\n",
 458     "mailbox.logout()"
 459    ]
 460   },
 461   {
 462    "cell_type": "code",
 463    "execution_count": 18,
 464    "metadata": {
 465     "collapsed": false
 466    },
 467    "outputs": [],
 468    "source": [
 469     "with conn:\n",
 470     "    with conn.cursor() as cur:\n",
 471     "        for report in dmarc_reports:\n",
 472     "            cur.execute('select id, report_metadata_report_id from reports where report_metadata_report_id = %s;', \n",
 473     "                [report.find('./report_metadata/report_id').text])\n",
 474     "            results = cur.fetchall()\n",
 475     "            if not results:\n",
 476     "                print('write', report.find('./report_metadata/report_id').text)\n",
 477     "                write_report(conn, cur, report)"
 478    ]
 479   },
 480   {
 481    "cell_type": "code",
 482    "execution_count": 19,
 483    "metadata": {
 484     "collapsed": true
 485    },
 486    "outputs": [],
 487    "source": [
 488     "conn.close()"
 489    ]
 490   }
 491  ],
 492  "metadata": {
 493   "kernelspec": {
 494    "display_name": "Python 3",
 495    "language": "python",
 496    "name": "python3"
 497   },
 498   "language_info": {
 499    "codemirror_mode": {
 500     "name": "ipython",
 501     "version": 3
 502    },
 503    "file_extension": ".py",
 504    "mimetype": "text/x-python",
 505    "name": "python",
 506    "nbconvert_exporter": "python",
 507    "pygments_lexer": "ipython3",
 508    "version": "3.4.3+"
 509   }
 510  },
 511  "nbformat": 4,
 512  "nbformat_minor": 0
 513 }