1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
|
# smlmp_common.py: common functions used in smlmp
# Copyright (C) 2023 Andrew Yu <https://www.andrewyu.org/>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
from __future__ import annotations
from typing import Optional, Union, Any
import configparser
import os
import json
import time
import pathlib
import subprocess
import errno
import fcntl
import email
import email.policy
import smtplib
import re
import traceback
def get_config() -> configparser.ConfigParser:
config = configparser.ConfigParser()
if not config.read("/etc/smlmp.conf"): # if unreadable
raise FileNotFoundError("/etc/smlmp.conf")
config["general"]["administrator"] = config["general"]["administrator"].lower()
config["general"]["localname"] = config["general"]["localname"].lower()
config["general"]["domain"] = config["general"]["domain"].lower()
config["delivery agent"]["archiver_address"] = config["delivery agent"]["archiver_address"].lower()
return config
config = get_config()
if not config["general"]["web_root"].endswith("/"):
config["general"]["web_root"] += "/"
policy = email.policy.SMTP.clone(refold_source="none", mangle_from_=False, max_line_length=998, cte_type="8bit", raise_on_defect=True)
class SMLMPException(Exception):
report_subject = "SMLMP Exception"
class SMLMPCriticalError(SMLMPException):
report_subject = "SMLMP Critical Error"
class SMLMPInvalidConfiguration(SMLMPCriticalError):
report_subject = "SMLMP Invalid Configuration"
class SMLMPRecipientError(SMLMPException):
report_subject = "SMLMP Recipient Error"
class SendmailError(SMLMPRecipientError):
report_subject = "SMLMP Sendmail Error"
class SMLMPSenderError(SMLMPException):
report_subject = "SMLMP Sender Error"
class SMLMPParseError(SMLMPSenderError):
report_subject = "SMLMP Parse Error"
class SMLMPLockTimeout(SMLMPException):
report_subject = "SMLMP Lock Timeout"
def sendmail(
message: email.message.EmailMessage,
specified_recipients_only: bool = False,
extra_recipients: list[str] = [],
) -> None:
bounce_address = config["general"]["localname"] + config["general"]["recipient_delimiter"] + "bounces@" + config["general"]["domain"]
conn = smtplib.SMTP()
conn.connect(config["general"]["smtp_host"], int(config["general"]["smtp_port"]))
if specified_recipients_only:
recipients = extra_recipients
else:
recipients = extra_recipients + extract_recipient_addresses(message)
conn.sendmail(bounce_address, recipients, message.as_bytes(policy=policy))
conn.quit()
def tell_administrator(message: email.message.EmailMessage) -> None:
sendmail(message, specified_recipients_only=True, extra_recipients=[config["general"]["administrator"]])
def extract_recipient_addresses(message: email.message.EmailMessage) -> list[str]:
to_addresses = (
[address.username.lower() + "@" + address.domain.lower() for address in message["To"].addresses] if message["To"] else []
)
cc_addresses = (
[address.username.lower() + "@" + address.domain.lower() for address in message["CC"].addresses] if message["CC"] else []
)
return to_addresses + cc_addresses
def report_error(e: Exception) -> None:
new_message = email.message.EmailMessage(policy=policy)
new_message["Subject"] = e.__class__.__name__
new_message["From"] = config["general"]["localname"] + "@" + config["general"]["domain"]
new_message["To"] = config["general"]["administrator"]
new_message.set_content(''.join(traceback.format_exception(e))) # Python 3.10 or above
tell_administrator(new_message)
def parse_local_address(address: str) -> tuple[str, str, str]:
if config["general"]["recipient_delimiter"] in address:
list_name, remaining = address.split(config["general"]["recipient_delimiter"], 1)
extension, domain = remaining.rsplit("@", 1)
else:
list_name, domain = address.rsplit("@", 1)
extension = ""
return list_name.lower(), extension.lower(), domain.lower()
def parse_dkim_header(dkim_header: str) -> tuple[set[str], dict[str, str]]:
# Adapted from dkimpy, modified from original software
#
# This function, parse_dkim_header, is also covered under the following
# copyright and license. THIS LICENSE DOES NOT APPLY TO THE ENTIRE PROGRAM.
#
# Copyright (C) 2023 Andrew Yu <https://www.andrewyu.org/>
# Copyright (C) 2019 Scott Kitterman <scott@kitterman.com>
# Copyright (C) 2017 Gene Shuman <gene@valimail.com>
# Copyright (C) 2011, 2012 Stuart D. Gathman
#
# This software is provided 'as-is', without any express or implied
# warranty. In no event will the author be held liable for any damages
# arising from the use of this software.
#
# Permission is granted to anyone to use this software for any purpose,
# including commercial applications, and to alter it and redistribute it
# freely, subject to the following restrictions:
#
# 1. The origin of this software must not be misrepresented; you must not
# claim that you wrote the original software. If you use this software
# in a product, an acknowledgment in the product documentation would be
# appreciated but is not required.
# 2. Altered source versions must be plainly marked as such, and must not be
# misrepresented as being the original software.
# 3. This notice may not be removed or altered from any source distribution.
tags = {}
tag_specs = dkim_header.strip().split(";")
# Trailing semicolons are valid.
if not tag_specs[-1]:
tag_specs.pop()
for tag_spec in tag_specs:
try:
key, value = [x.strip() for x in tag_spec.split("=", 1)]
except ValueError:
raise SMLMPParseError("invalid tag spec", tag_spec)
if re.match(r"^[a-zA-Z](\w)*", key) is None:
raise SMLMPParseError("invalid tag spec", tag_spec)
if key in tags:
raise SMLMPParseError("duplicate tag", key)
tags[key] = value
dkim_include_headers = set([x.lower() for x in re.split(r"\s*:\s*", tags["h"])])
return dkim_include_headers, tags
# def read_db() -> None:
# initial_time = time.time()
# timeout = False
# while os.exists(config["general"]["database"] + ".write_lock"):
# if time.time() - initial_time < int(config["general"]["lock_timeout"]):
# raise SMLMPLockTimeout("Lock not released for more than lock_timeout")
# # now we hope that nobody immediatly locks the file again to write (which we can't detect) as we try to read it
# with open(config["general"]["database"], "r") as db_file:
# db = json.load(db_file)
# return db
def read_db() -> dict[str, Any]:
with open(config["general"]["database"], "r") as db_file:
fcntl.flock(db_file, fcntl.LOCK_SH) # | fcntl.LOCK_NB
db = json.load(db_file)
fcntl.flock(db_file, fcntl.LOCK_UN)
assert type(db) is dict
return db
# BLOCKS until it could acquire the lock - also we're blocking two reads from happening together
|
