dotfiles/config/scripts/smtpdummy

189 lines
5.1 KiB
Plaintext
Raw Normal View History

2020-06-28 11:23:02 +02:00
#!/usr/bin/env python3
2020-12-27 14:20:44 +01:00
import base64
import colorama
import configargparse
2020-06-28 11:23:02 +02:00
import datetime
import email.utils
2020-12-27 14:20:44 +01:00
import io
2020-06-28 11:23:02 +02:00
import pprint
2020-12-27 14:20:44 +01:00
import subprocess
import sys
2020-06-28 11:23:02 +02:00
if __name__ == "__main__":
2020-12-27 14:20:44 +01:00
parser = configargparse.ArgParser(
2020-06-28 11:23:02 +02:00
description="Generate SMTP messages to send a mail"
)
now = datetime.datetime.now()
now_email = email.utils.formatdate(now.timestamp(), True)
2020-12-27 14:20:44 +01:00
parser.add_argument("-o", "--origin", env_var="ORIGIN", default="localhost")
parser.add_argument(
"-d", "--destination", env_var="DESTINATION", default="localhost"
)
parser.add_argument("-p", "--port", env_var="PORT", default=25)
2020-06-28 11:23:02 +02:00
parser.add_argument(
2020-12-27 14:20:44 +01:00
"-S", "--security", env_var="SECURITY", choices=["plain", "ssl", "starttls"], default='plain'
2020-06-28 11:23:02 +02:00
)
2020-12-27 14:20:44 +01:00
parser.add_argument("-l", "--helo", env_var="HELO")
parser.add_argument(
"-s", "--sender", env_var="SENDER", default="geoffrey@frogeye.fr"
)
parser.add_argument(
"-r", "--receiver", env_var="RECEIVER", default="geoffrey@frogeye.fr"
)
# parser.add_argument("-a", "--auth", env_var="AUTH", default="PLAIN")
parser.add_argument("-u", "--user", env_var="MUSER")
parser.add_argument("-w", "--password", env_var="PASSWORD")
parser.add_argument("-f", "--from", env_var="FROM")
parser.add_argument("-t", "--to", env_var="TO")
parser.add_argument(
"-j",
"--subject",
env_var="SUBJECT",
default=f"Test message {now.strftime('%H:%M:%S')}",
)
parser.add_argument("-b", "--body", env_var="BODY", default="")
parser.add_argument("-g", "--gtube", env_var="GTUBE", action="store_true")
parser.add_argument("-m", "--me", env_var="ME", default="Geoffrey")
parser.add_argument("-y", "--dryrun", env_var="DRYRUN", action="store_true")
parser.add_argument("-q", "--quiet", env_var="QUIET", action="store_true")
2020-06-28 11:23:02 +02:00
args = parser.parse_args()
2020-12-27 14:20:44 +01:00
# Default values
if args.helo is None:
args.helo = args.origin
if getattr(args, "from") is None:
setattr(args, "from", args.sender)
if args.to is None:
args.to = args.receiver
if args.password:
password = args.password
args.password = '********'
# Transmission content
2020-06-28 11:23:02 +02:00
2020-08-08 11:19:48 +02:00
gtube = ""
if args.gtube:
gtube = """
XJS*C4JDBQADN1.NSBN3*2IDNEN*GTUBE-STANDARD-ANTI-UBE-TEST-EMAIL*C.34X"""
body = ""
if args.body:
body = f"\n\n{args.body}"
2020-06-28 11:23:02 +02:00
text = f"""Date: {now_email}
2020-12-27 14:20:44 +01:00
From: {getattr(args, 'from')}
2020-06-28 11:23:02 +02:00
Subject: {args.subject}
2020-12-27 14:20:44 +01:00
To: {args.to}
2020-06-28 11:23:02 +02:00
Hello there,
This is a test message, generated from a template.
2020-08-08 11:19:48 +02:00
If you didn't expect to see this message, please contact {args.me}.{gtube}{body}
2020-06-28 11:23:02 +02:00
Greetings,
Input arguments:
2020-12-27 14:20:44 +01:00
{pprint.pformat(args.__dict__, indent=4)}
2020-06-28 11:23:02 +02:00
--
{args.me}
."""
2020-12-27 14:20:44 +01:00
# Transmission setup
cmd = ["ssh", args.origin]
if args.security == "plain":
cmd += ["socat", "-", f"tcp:{args.destination}:{args.port}"]
elif args.security == "ssl":
cmd += ["socat", "-", f"openssl:{args.destination}:{args.port}"]
elif args.security == "starttls":
cmd += [
"openssl",
"s_client",
"-starttls",
"smtp",
"-crlf",
"-connect",
f"{args.destination}:{args.port}",
"-quiet",
]
if not args.quiet:
print(colorama.Fore.MAGENTA + f"# {' '.join(cmd)}" + colorama.Fore.RESET)
if not args.dryrun:
p = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL,
)
def recv() -> None:
if args.dryrun:
return
assert isinstance(p.stdout, io.BufferedReader)
next = True
while next:
line = p.stdout.readline()
code = int(line[:3])
success = code < 400
color = colorama.Fore.GREEN if success else colorama.Fore.RED
if not args.quiet:
print(color + f"< {line[:-1].decode()}" + colorama.Fore.RESET)
next = line[3] == b"-"[0]
if not next and not success:
send("QUIT") # TODO Can loop if QUIT fails
sys.exit(1)
def send(command: str) -> None:
if not args.quiet:
print(colorama.Fore.BLUE + f"> {command}" + colorama.Fore.RESET)
if args.dryrun:
return
assert isinstance(p.stdin, io.BufferedWriter)
cmd = command.encode() + b"\n"
p.stdin.write(cmd)
p.stdin.flush()
recv()
# Transmission
if args.security != 'starttls':
recv()
send(f"EHLO {args.helo}")
if args.user:
encoded = base64.b64encode(
args.user.encode()
+ b"\x00"
+ args.user.encode()
+ b"\x00"
+ password.encode()
).decode()
send(f"AUTH PLAIN {encoded}")
send(f"MAIL FROM: <{args.sender}>")
send(f"RCPT TO: <{args.receiver}>")
send("DATA")
send(text)
send("QUIT")
sys.exit(0)
2020-06-28 11:23:02 +02:00
# For reference:
2020-12-27 14:20:44 +01:00
# send("RSET")
# send("VRFY")
# send("NOOP")
# send("QUIT")