Added config file for the Ruff Python linter and fixed additional errors

management/backup.py

@@ -14,6 +14,7 @@
 from exclusiveprocess import Lock
 
 from utils import load_environment, shell, wait_for_service
+import operator
 
 def backup_status(env):
 	# If backups are disabled, return no status.
@@ -91,7 +92,7 @@ def parse_line(line):
 
 	# Ensure the rows are sorted reverse chronologically.
 	# This is relied on by should_force_full() and the next step.
-	backups = sorted(backups.values(), key = lambda b : b["date"], reverse=True)
+	backups = sorted(backups.values(), key = operator.itemgetter("date"), reverse=True)
 
 	# Get the average size of incremental backups, the size of the
 	# most recent full backup, and the date of the most recent
@@ -177,10 +178,9 @@ def should_force_full(config, env):
 				if dateutil.parser.parse(bak["date"]) + datetime.timedelta(days=config["min_age_in_days"]*10+1) < datetime.datetime.now(dateutil.tz.tzlocal()):
 					return True
 			return False
-	else:
-		# If we got here there are no (full) backups, so make one.
-		# (I love for/else blocks. Here it's just to show off.)
-		return True
+	# If we got here there are no (full) backups, so make one.
+	# (I love for/else blocks. Here it's just to show off.)
+	return True
 
 def get_passphrase(env):
 	# Get the encryption passphrase. secret_key.txt is 2048 random
@@ -236,7 +236,7 @@ def get_duplicity_additional_args(env):
 			f"--ssh-options='-i /root/.ssh/id_rsa_miab -p {port}'",
 			f"--rsync-options='-e \"/usr/bin/ssh -oStrictHostKeyChecking=no -oBatchMode=yes -p {port} -i /root/.ssh/id_rsa_miab\"'",
 		]
-	elif get_target_type(config) == 's3':
+	if get_target_type(config) == 's3':
 		# See note about hostname in get_duplicity_target_url.
 		# The region name, which is required by some non-AWS endpoints,
 		# is saved inside the username portion of the URL.
@@ -447,7 +447,7 @@ def list_target_files(config):
 	if target.scheme == "file":
 		return [(fn, os.path.getsize(os.path.join(target.path, fn))) for fn in os.listdir(target.path)]
 
-	elif target.scheme == "rsync":
+	if target.scheme == "rsync":
 		rsync_fn_size_re = re.compile(r'.*    ([^ ]*) [^ ]* [^ ]* (.*)')
 		rsync_target = '{host}:{path}'
 
@@ -463,9 +463,8 @@ def list_target_files(config):
 
 		target_path = target.path
 		if not target_path.endswith('/'):
-			target_path = target_path + '/'
-		if target_path.startswith('/'):
-			target_path = target_path[1:]
+			target_path += "/"
+		target_path = target_path.removeprefix('/')
 
 		rsync_command = [ 'rsync',
 					'-e',
@@ -485,21 +484,20 @@ def list_target_files(config):
 				if match:
 					ret.append( (match.groups()[1], int(match.groups()[0].replace(',',''))) )
 			return ret
+		if 'Permission denied (publickey).' in listing:
+			reason = "Invalid user or check you correctly copied the SSH key."
+		elif 'No such file or directory' in listing:
+			reason = f"Provided path {target_path} is invalid."
+		elif 'Network is unreachable' in listing:
+			reason = f"The IP address {target.hostname} is unreachable."
+		elif 'Could not resolve hostname' in listing:
+			reason = f"The hostname {target.hostname} cannot be resolved."
 		else:
-			if 'Permission denied (publickey).' in listing:
-				reason = "Invalid user or check you correctly copied the SSH key."
-			elif 'No such file or directory' in listing:
-				reason = f"Provided path {target_path} is invalid."
-			elif 'Network is unreachable' in listing:
-				reason = f"The IP address {target.hostname} is unreachable."
-			elif 'Could not resolve hostname' in listing:
-				reason = f"The hostname {target.hostname} cannot be resolved."
-			else:
-				reason = ("Unknown error."
-						"Please check running 'management/backup.py --verify'"
-						"from mailinabox sources to debug the issue.")
-			msg = f"Connection to rsync host failed: {reason}"
-			raise ValueError(msg)
+			reason = ("Unknown error."
+					"Please check running 'management/backup.py --verify'"
+					"from mailinabox sources to debug the issue.")
+		msg = f"Connection to rsync host failed: {reason}"
+		raise ValueError(msg)
 
 	elif target.scheme == "s3":
 		import boto3.s3
@@ -602,8 +600,7 @@ def get_backup_config(env, for_save=False, for_ui=False):
 	# authentication details. The user will have to re-enter it.
 	if for_ui:
 		for field in ("target_user", "target_pass"):
-			if field in config:
-				del config[field]
+			config.pop(field, None)
 
 	# helper fields for the admin
 	config["file_target_directory"] = os.path.join(backup_root, 'encrypted')

management/daemon.py

@@ -92,12 +92,11 @@ def newview(*args, **kwargs):
 		if request.headers.get('Accept') in {None, "", "*/*"}:
 			# Return plain text output.
 			return Response(error+"\n", status=status, mimetype='text/plain', headers=headers)
-		else:
-			# Return JSON output.
-			return Response(json.dumps({
-				"status": "error",
-				"reason": error,
-				})+"\n", status=status, mimetype='application/json', headers=headers)
+		# Return JSON output.
+		return Response(json.dumps({
+			"status": "error",
+			"reason": error,
+			})+"\n", status=status, mimetype='application/json', headers=headers)
 
 	return newview
 
@@ -147,13 +146,12 @@ def login():
 				"status": "missing-totp-token",
 				"reason": str(e),
 			})
-		else:
-			# Log the failed login
-			log_failed_login(request)
-			return json_response({
-				"status": "invalid",
-				"reason": str(e),
-			})
+		# Log the failed login
+		log_failed_login(request)
+		return json_response({
+			"status": "invalid",
+			"reason": str(e),
+		})
 
 	# Return a new session for the user.
 	resp = {
@@ -185,8 +183,7 @@ def logout():
 def mail_users():
 	if request.args.get("format", "") == "json":
 		return json_response(get_mail_users_ex(env, with_archived=True))
-	else:
-		return "".join(x+"\n" for x in get_mail_users(env))
+	return "".join(x+"\n" for x in get_mail_users(env))
 
 @app.route('/mail/users/add', methods=['POST'])
 @authorized_personnel_only
@@ -233,8 +230,7 @@ def mail_user_privs_remove():
 def mail_aliases():
 	if request.args.get("format", "") == "json":
 		return json_response(get_mail_aliases_ex(env))
-	else:
-		return "".join(address+"\t"+receivers+"\t"+(senders or "")+"\n" for address, receivers, senders, auto in get_mail_aliases(env))
+	return "".join(address+"\t"+receivers+"\t"+(senders or "")+"\n" for address, receivers, senders, auto in get_mail_aliases(env))
 
 @app.route('/mail/aliases/add', methods=['POST'])
 @authorized_personnel_only
@@ -354,7 +350,7 @@ def dns_set_record(qname, rtype="A"):
 			# Get the existing records matching the qname and rtype.
 			return dns_get_records(qname, rtype)
 
-		elif request.method in {"POST", "PUT"}:
+		if request.method in {"POST", "PUT"}:
 			# There is a default value for A/AAAA records.
 			if rtype in {"A", "AAAA"} and value == "":
 				value = request.environ.get("HTTP_X_FORWARDED_FOR") # normally REMOTE_ADDR but we're behind nginx as a reverse proxy
@@ -512,8 +508,8 @@ def totp_post_disable():
 		return (str(e), 400)
 	if result: # success
 		return "OK"
-	else: # error
-		return ("Invalid user or MFA id.", 400)
+	# error
+	return ("Invalid user or MFA id.", 400)
 
 # WEB
 
@@ -597,8 +593,7 @@ def needs_reboot():
 	from status_checks import is_reboot_needed_due_to_package_installation
 	if is_reboot_needed_due_to_package_installation():
 		return json_response(True)
-	else:
-		return json_response(False)
+	return json_response(False)
 
 @app.route('/system/reboot', methods=["POST"])
 @authorized_personnel_only
@@ -607,8 +602,7 @@ def do_reboot():
 	from status_checks import is_reboot_needed_due_to_package_installation
 	if is_reboot_needed_due_to_package_installation():
 		return utils.shell("check_output", ["/sbin/shutdown", "-r", "now"], capture_stderr=True)
-	else:
-		return "No reboot is required, so it is not allowed."
+	return "No reboot is required, so it is not allowed."
 
 
 @app.route('/system/backup/status')
@@ -670,8 +664,7 @@ def check_request_cookie_for_admin_access():
 	if not session: return False
 	privs = get_mail_user_privileges(session["email"], env)
 	if not isinstance(privs, list): return False
-	if "admin" not in privs: return False
-	return True
+	return "admin" in privs
 
 def authorized_personnel_only_via_cookie(f):
 	@wraps(f)
@@ -719,7 +712,7 @@ def munin_cgi(filename):
 
 	query_str = request.query_string.decode("utf-8", 'ignore')
 
-	env = {'PATH_INFO': '/%s/' % filename, 'REQUEST_METHOD': 'GET', 'QUERY_STRING': query_str}
+	env = {'PATH_INFO': f'/{filename}/', 'REQUEST_METHOD': 'GET', 'QUERY_STRING': query_str}
 	code, binout = utils.shell('check_output',
 							   COMMAND.split(" ", 5),
 							   # Using a maxsplit of 5 keeps the last arguments together

management/dns_update.py

@@ -11,7 +11,6 @@
 
 from utils import shell, load_env_vars_from_file, safe_domain_name, sort_domains, get_ssh_port
 from ssl_certificates import get_ssl_certificates, check_certificate
-import contextlib
 
 # From https://stackoverflow.com/questions/3026957/how-to-validate-a-domain-name-using-regex-php/16491074#16491074
 # This regular expression matches domain names according to RFCs, it also accepts fqdn with an leading dot,
@@ -124,8 +123,7 @@ def do_dns_update(env, force=False):
 	if len(updated_domains) == 0:
 		# if nothing was updated (except maybe OpenDKIM's files), don't show any output
 		return ""
-	else:
-		return "updated DNS: " + ",".join(updated_domains) + "\n"
+	return "updated DNS: " + ",".join(updated_domains) + "\n"
 
 ########################################################################
 
@@ -187,7 +185,7 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True)
 	# is managed outside of the box.
 	if is_zone:
 		# Obligatory NS record to ns1.PRIMARY_HOSTNAME.
-		records.append((None,  "NS",  "ns1.%s." % env["PRIMARY_HOSTNAME"], False))
+		records.append((None,  "NS",  "ns1.{}.".format(env["PRIMARY_HOSTNAME"]), False))
 
 		# NS record to ns2.PRIMARY_HOSTNAME or whatever the user overrides.
 		# User may provide one or more additional nameservers
@@ -254,16 +252,16 @@ def has_rec(qname, rtype, prefix=None):
 	# was set. So set has_rec_base to a clone of the current set of DNS settings, and don't update
 	# during this process.
 	has_rec_base = list(records)
-	a_expl = "Required. May have a different value. Sets the IP address that %s resolves to for web hosting and other services besides mail. The A record must be present but its value does not affect mail delivery." % domain
+	a_expl = f"Required. May have a different value. Sets the IP address that {domain} resolves to for web hosting and other services besides mail. The A record must be present but its value does not affect mail delivery."
 	if domain_properties[domain]["auto"]:
 		if domain.startswith(("ns1.", "ns2.")): a_expl = False # omit from 'External DNS' page since this only applies if box is its own DNS server
-		if domain.startswith("www."): a_expl = "Optional. Sets the IP address that %s resolves to so that the box can provide a redirect to the parent domain." % domain
+		if domain.startswith("www."): a_expl = f"Optional. Sets the IP address that {domain} resolves to so that the box can provide a redirect to the parent domain."
 		if domain.startswith("mta-sts."): a_expl = "Optional. MTA-STS Policy Host serving /.well-known/mta-sts.txt."
 		if domain.startswith("autoconfig."): a_expl = "Provides email configuration autodiscovery support for Thunderbird Autoconfig."
 		if domain.startswith("autodiscover."): a_expl = "Provides email configuration autodiscovery support for Z-Push ActiveSync Autodiscover."
 	defaults = [
 		(None,  "A",    env["PUBLIC_IP"], a_expl),
-		(None,  "AAAA", env.get('PUBLIC_IPV6'), "Optional. Sets the IPv6 address that %s resolves to, e.g. for web hosting. (It is not necessary for receiving mail on this domain.)" % domain),
+		(None,  "AAAA", env.get('PUBLIC_IPV6'), f"Optional. Sets the IPv6 address that {domain} resolves to, e.g. for web hosting. (It is not necessary for receiving mail on this domain.)"),
 	]
 	for qname, rtype, value, explanation in defaults:
 		if value is None or value.strip() == "": continue # skip IPV6 if not set
@@ -281,13 +279,13 @@ def has_rec(qname, rtype, prefix=None):
 	if domain_properties[domain]["mail"]:
 		# The MX record says where email for the domain should be delivered: Here!
 		if not has_rec(None, "MX", prefix="10 "):
-			records.append((None,  "MX",  "10 %s." % env["PRIMARY_HOSTNAME"], "Required. Specifies the hostname (and priority) of the machine that handles @%s mail." % domain))
+			records.append((None,  "MX",  "10 {}.".format(env["PRIMARY_HOSTNAME"]), f"Required. Specifies the hostname (and priority) of the machine that handles @{domain} mail."))
 
 		# SPF record: Permit the box ('mx', see above) to send mail on behalf of
 		# the domain, and no one else.
 		# Skip if the user has set a custom SPF record.
 		if not has_rec(None, "TXT", prefix="v=spf1 "):
-			records.append((None,  "TXT", 'v=spf1 mx -all', "Recommended. Specifies that only the box is permitted to send @%s mail." % domain))
+			records.append((None,  "TXT", 'v=spf1 mx -all', f"Recommended. Specifies that only the box is permitted to send @{domain} mail."))
 
 		# Append the DKIM TXT record to the zone as generated by OpenDKIM.
 		# Skip if the user has set a DKIM record already.
@@ -296,12 +294,12 @@ def has_rec(qname, rtype, prefix=None):
 			m = re.match(r'(\S+)\s+IN\s+TXT\s+\( ((?:"[^"]+"\s+)+)\)', orf.read(), re.S)
 			val = "".join(re.findall(r'"([^"]+)"', m.group(2)))
 			if not has_rec(m.group(1), "TXT", prefix="v=DKIM1; "):
-				records.append((m.group(1), "TXT", val, "Recommended. Provides a way for recipients to verify that this machine sent @%s mail." % domain))
+				records.append((m.group(1), "TXT", val, f"Recommended. Provides a way for recipients to verify that this machine sent @{domain} mail."))
 
 		# Append a DMARC record.
 		# Skip if the user has set a DMARC record already.
 		if not has_rec("_dmarc", "TXT", prefix="v=DMARC1; "):
-			records.append(("_dmarc", "TXT", 'v=DMARC1; p=quarantine;', "Recommended. Specifies that mail that does not originate from the box but claims to be from @%s or which does not have a valid DKIM signature is suspect and should be quarantined by the recipient's mail system." % domain))
+			records.append(("_dmarc", "TXT", 'v=DMARC1; p=quarantine;', f"Recommended. Specifies that mail that does not originate from the box but claims to be from @{domain} or which does not have a valid DKIM signature is suspect and should be quarantined by the recipient's mail system."))
 
 	if domain_properties[domain]["user"]:
 		# Add CardDAV/CalDAV SRV records on the non-primary hostname that points to the primary hostname
@@ -364,9 +362,9 @@ def has_rec(qname, rtype, prefix=None):
 			# Mark this domain as not sending mail with hard-fail SPF and DMARC records.
 			d = (qname+"." if qname else "") + domain
 			if not has_rec(qname, "TXT", prefix="v=spf1 "):
-				records.append((qname,  "TXT", 'v=spf1 -all', "Recommended. Prevents use of this domain name for outbound mail by specifying that no servers are valid sources for mail from @%s. If you do send email from this domain name you should either override this record such that the SPF rule does allow the originating server, or, take the recommended approach and have the box handle mail for this domain (simply add any receiving alias at this domain name to make this machine treat the domain name as one of its mail domains)." % d))
+				records.append((qname,  "TXT", 'v=spf1 -all', f"Recommended. Prevents use of this domain name for outbound mail by specifying that no servers are valid sources for mail from @{d}. If you do send email from this domain name you should either override this record such that the SPF rule does allow the originating server, or, take the recommended approach and have the box handle mail for this domain (simply add any receiving alias at this domain name to make this machine treat the domain name as one of its mail domains)."))
 			if not has_rec("_dmarc" + ("."+qname if qname else ""), "TXT", prefix="v=DMARC1; "):
-				records.append(("_dmarc" + ("."+qname if qname else ""), "TXT", 'v=DMARC1; p=reject;', "Recommended. Prevents use of this domain name for outbound mail by specifying that the SPF rule should be honoured for mail from @%s." % d))
+				records.append(("_dmarc" + ("."+qname if qname else ""), "TXT", 'v=DMARC1; p=reject;', f"Recommended. Prevents use of this domain name for outbound mail by specifying that the SPF rule should be honoured for mail from @{d}."))
 
 			# And with a null MX record (https://explained-from-first-principles.com/email/#null-mx-record)
 			if not has_rec(qname, "MX"):
@@ -592,7 +590,7 @@ def get_dns_zonefile(zone, env):
 		if zone == domain:
 			break
 	else:
-		raise ValueError("%s is not a domain name that corresponds to a zone." % zone)
+		raise ValueError(f"{zone} is not a domain name that corresponds to a zone.")
 
 	nsd_zonefile = "/etc/nsd/zones/" + fn
 	with open(nsd_zonefile, encoding="utf-8") as f:
@@ -617,8 +615,8 @@ def write_nsd_conf(zonefiles, additional_records, env):
 		# and, if not a subnet, notifies to them.
 		for ipaddr in get_secondary_dns(additional_records, mode="xfr"):
 			if "/" not in ipaddr:
-				nsdconf += "\n\tnotify: %s NOKEY" % (ipaddr)
-			nsdconf += "\n\tprovide-xfr: %s NOKEY\n" % (ipaddr)
+				nsdconf += f"\n\tnotify: {ipaddr} NOKEY"
+			nsdconf += f"\n\tprovide-xfr: {ipaddr} NOKEY\n"
 
 	# Check if the file is changing. If it isn't changing,
 	# return False to flag that no change was made.
@@ -898,7 +896,7 @@ def set_custom_dns_record(qname, rtype, value, action, env):
 	else:
 		# No match.
 		if qname != "_secondary_nameserver":
-			raise ValueError("%s is not a domain name or a subdomain of a domain name managed by this box." % qname)
+			raise ValueError(f"{qname} is not a domain name or a subdomain of a domain name managed by this box.")
 
 	# validate rtype
 	rtype = rtype.upper()
@@ -919,7 +917,7 @@ def set_custom_dns_record(qname, rtype, value, action, env):
 
 			# ensure value has a trailing dot
 			if not value.endswith("."):
-				value = value + "."
+				value += "."
 
 			if not re.search(DOMAIN_RE, value):
 				msg = "Invalid value."
@@ -928,7 +926,7 @@ def set_custom_dns_record(qname, rtype, value, action, env):
 			# anything goes
 			pass
 		else:
-			raise ValueError("Unknown record type '%s'." % rtype)
+			raise ValueError(f"Unknown record type '{rtype}'.")
 
 	# load existing config
 	config = list(get_custom_dns_config(env))
@@ -1039,7 +1037,7 @@ def set_secondary_dns(hostnames, env):
 					try:
 						resolver.resolve(item, "AAAA")
 					except (dns.resolver.NoNameservers, dns.resolver.NXDOMAIN, dns.resolver.NoAnswer, dns.resolver.Timeout):
-						raise ValueError("Could not resolve the IP address of %s." % item)
+						raise ValueError(f"Could not resolve the IP address of {item}.")
 			else:
 				# Validate IP address.
 				try:
@@ -1048,7 +1046,7 @@ def set_secondary_dns(hostnames, env):
 					else:
 						ipaddress.ip_address(item[4:]) # raises a ValueError if there's a problem
 				except ValueError:
-					raise ValueError("'%s' is not an IPv4 or IPv6 address or subnet." % item[4:])
+					raise ValueError(f"'{item[4:]}' is not an IPv4 or IPv6 address or subnet.")
 
 		# Set.
 		set_custom_dns_record("_secondary_nameserver", "A", " ".join(hostnames), "set", env)