11 "import configparser\n",
16 "import xml.etree.ElementTree\n",
30 "def fetch_msg(num):\n",
31 " return mailbox.uid('FETCH', num, '(RFC822)')[1][0][1]"
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)"
57 "def xml_of(message):\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",
64 " reports = [xml_of_part(message)]\n",
76 "def extract_report(msg):\n",
77 " pmsg = email.message_from_bytes(msg)\n",
78 " return xml_of(pmsg)"
89 "def maybe_strip(text):\n",
91 " return text.strip()\n",
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'}}"
168 "execution_count": 8,
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",
180 " for f in [f for f in field_maps if field_maps[f]['pg_table'] == table_name]:\n",
183 " fp = f.format(i)\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",
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"
208 "execution_count": 9,
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",
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",
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()"
237 "execution_count": 10,
243 "def write_report_old(connection, cursor, report):\n",
245 " field_names = []\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",
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",
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",
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",
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()"
291 "execution_count": 11,
302 "execution_count": 11,
304 "output_type": "execute_result"
308 "config = configparser.ConfigParser()\n",
309 "config.read('dmarc.ini')"
314 "execution_count": 12,
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'])"
328 "execution_count": 13,
336 "datetime.datetime(2016, 3, 30, 23, 59, 59)"
339 "execution_count": 13,
341 "output_type": "execute_result"
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",
355 "execution_count": 14,
366 "execution_count": 14,
368 "output_type": "execute_result"
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)"
381 "execution_count": 15,
389 "('SINCE 28-Mar-2016', 'OK', [b'172 173 174 175 176 177 178 179 180'])"
392 "execution_count": 15,
394 "output_type": "execute_result"
398 "if most_recent_date:\n",
399 " mails_from = \"SINCE \" + (most_recent_date - datetime.timedelta(days=2)).strftime(\"%d-%b-%Y\")\n",
401 " mails_from = \"ALL\"\n",
402 "resp, nums = mailbox.uid('SEARCH', None, mails_from)\n",
403 "mails_from, resp, nums"
408 "execution_count": 16,
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']"
427 "execution_count": 16,
429 "output_type": "execute_result"
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]"
440 "execution_count": 17,
448 "('BYE', [b'Logging out'])"
451 "execution_count": 17,
453 "output_type": "execute_result"
463 "execution_count": 18,
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)"
482 "execution_count": 19,
494 "display_name": "Python 3",
495 "language": "python",
503 "file_extension": ".py",
504 "mimetype": "text/x-python",
506 "nbconvert_exporter": "python",
507 "pygments_lexer": "ipython3",