Tandem/spf-dkim-dmarc-demo
2
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

send_email.py : ajouter une sortie JSON

Browse Source 
Faire un sorte que le script puisse fonctionner en mode non interactif :
dans ce cas, sa sortie sur la console est du JSON. Cela permet ensuite à
une éventuelle API REST de facilement se pluguer dessus.
This commit is contained in:
Marc van der Wal 2023-10-25 15:50:24 +02:00
parent cacba346f6
commit 1e14402a92
1 changed files with 83 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

110
attacker/scripts/send_email.py
View File

@ -1,8 +1,10 @@
#!/usr/bin/env python3
from argparse import ArgumentParser
from email.message import EmailMessage
import email.utils
import importlib.resources
import json
import pathlib
import re
import smtplib
@ -61,15 +63,18 @@ custom_theme = Theme({
"data_display_name": "medium_purple1",
"comment": "#777777",
"failure": "bold red",
"success": "green"
"success": "green",
"text": "default"
})
console = Console(highlighter=None, theme=custom_theme)
class SMTP(smtplib.SMTP):
def __init__(self, **kwargs):
def __init__(self, output, **kwargs):
super().__init__(**kwargs)
self.highlighter = SMTPHighlighter()
self.log = [];
self.output = output
def trace(self, s, direction):
if direction not in ['in', 'out']:
@ -80,26 +85,32 @@ class SMTP(smtplib.SMTP):
if isinstance(s, bytes):
s = s.decode('utf-8')
lines = [self.highlighter(rich.text.Text(line)) for line in s.splitlines()]
lines = [{'class': 'text', 'text': text} for text in s.splitlines()]
# lines = [self.highlighter(rich.text.Text(line)) for line in s.splitlines()]
if len(lines) > max_lines:
suppressed_count = len(lines[max_lines:-1])
suppressed_text_obj = rich.text.Text(f"({suppressed_count} ")
if suppressed_count > 1:
suppressed_text_obj.append("lignes omises")
suppressed_text = f"{suppressed_count} lignes omises"
else:
suppressed_text_obj.append("ligne omise")
suppressed_text_obj.append(")")
suppressed_text_obj.stylize("comment")
suppressed_text = f"{suppressed_count} ligne omise"
suppressed_obj = {'class': 'comment', 'text': suppressed_text}
lines = lines[:max_lines] + [suppressed_text_obj] + lines[-1:]
lines = lines[:max_lines] + [suppressed_obj] + lines[-1:]
for i, l in enumerate(lines):
prefix = arrows[direction] if i == 0 else " "
console.print(f" {prefix} ", end=None)
console.print(l)
# Print the text on the console
if self.output:
for i, l in enumerate(lines):
prefix = arrows[direction] if i == 0 else " "
console.print(f" {prefix} ", end=None)
console.print(l['text'], style=l['class'])
if direction == 'in':
console.print()
if direction == 'in':
console.print()
# Save it internally too
self.log.append({'direction': direction, 'lines': lines})
def putcmd(self, cmd, args=''):
super().putcmd(cmd.upper(), args)
@ -119,7 +130,9 @@ class SMTP(smtplib.SMTP):
return errcode, errmsg
def connect(self, *args, **kwargs):
console.print(" * Connexion établie au serveur", style="comment")
self.log.append({'direction': 'comment',
'lines': [{'class': 'comment',
'text': 'Connexion établie au serveur'}]})
return super().connect(*args, **kwargs)
def ehlo(self, *args, **kwargs):
@ -218,21 +231,20 @@ def ask_template():
def send_email(helo, envelope_from, email):
def send_email(helo, envelope_from, email, output=True):
data = email.as_string().encode('utf-8')
try:
with SMTP(local_hostname=helo) as smtp:
with SMTP(local_hostname=helo, output=output) as smtp:
try:
smtp.connect(VICTIM_MX)
smtp.sendmail(envelope_from, VICTIM_ADDRESS, data)
except smtplib.SMTPException:
console.print(" [failure]✘[/] Message [failure]rejeté[/] par le serveur SMTP")
except:
raise
else:
console.print(" [success]✔[/] Message [success]accepté[/] par le serveur SMTP")
return {'outcome': 'success', 'log': smtp.log}
except smtplib.SMTPException:
return {'outcome': 'failure', 'log': smtp.log}
except:
raise
def main():
def interactive_main():
try:
console.print(
rich.panel.Panel(
@ -256,7 +268,12 @@ def main():
email = generate_email(template)
signed_email = add_dkim_signature(email)
send_email(helo, envelope_from, signed_email)
results = send_email(helo, envelope_from, signed_email, output=True)
if results['outcome'] == 'success':
console.print(" [success]✔[/] Message [success]accepté[/] par le serveur SMTP")
elif results['outcome'] == 'failure':
console.print(" [failure]✘[/] Message [failure]rejeté[/] par le serveur SMTP")
except KeyboardInterrupt:
console.print()
pass
@ -264,5 +281,44 @@ def main():
raise
def main():
parser = ArgumentParser()
parser.add_argument('--non-interactive', default=False, action='store_true')
parser.add_argument('--get-config', default=False, action='store_true')
parser.add_argument('--template')
parser.add_argument('--replace-rfc5321-mail-from', default=False, action='store_true')
parser.add_argument('--helo', default=DEFAULT_HELO)
args = parser.parse_args()
if args.non_interactive:
if args.get_config:
config = {
'default_helo': DEFAULT_HELO,
'my_mail_from': ATTACKER_MAIL_FROM,
'templates': [{'id': t.name,
'name': t.nice_name,
'from_name': re.match('(.*) <(.*)>', t.msg['From']).group(1),
'from_address': re.match('(.*) <(.*)>', t.msg['From']).group(2)}
for t in spoof_templates.spoof_templates()]
}
print(json.dumps(config))
else:
templates = {t.name: t for t in spoof_templates.spoof_templates()}
template = templates[args.template]
rfc5322from, = re.match('.* <(.*)>', template.msg['From']).groups()
if args.replace_rfc5321_mail_from:
envelope_from = ATTACKER_MAIL_FROM
else:
envelope_from = rfc5322from
email = generate_email(template)
results = send_email(args.helo, envelope_from, email, output=False)
print(json.dumps(results))
else:
interactive_main()
if __name__ == '__main__':
main()