From 544cce3cdcdd328911fcde8425ba5983cd4ce45e Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Sat, 27 Jan 2024 08:32:32 -0800 Subject: [PATCH 01/37] Fixed syntax in readable_bash.py. --- tools/readable_bash.py | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 66f6196c..eb439397 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -125,13 +125,13 @@ def generate_documentation(): parser = Source.parser() with open("setup/start.sh", "r") as start_file: - for line in start_file: - try: - fn = parser.parse_string(line).filename() - except: - continue - if fn in ("setup/start.sh", "setup/preflight.sh", "setup/questions.sh", "setup/firstuser.sh", "setup/management.sh"): - continue + for line in start_file: + try: + fn = parser.parse_string(line).filename() + except: + continue + if fn in ("setup/start.sh", "setup/preflight.sh", "setup/questions.sh", "setup/firstuser.sh", "setup/management.sh"): + continue import sys print(fn, file=sys.stderr) From 698e8ffc72417017fa5ef45ab75e5dc98f266d21 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 8 Jan 2025 03:10:27 -0800 Subject: [PATCH 02/37] Added Ruff config for Python code. --- pyproject.toml | 69 ++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 69 insertions(+) create mode 100644 pyproject.toml diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 00000000..5401faef --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,69 @@ +[tool.ruff] +line-length = 320 # https://github.com/astral-sh/ruff/issues/8106 +indent-width = 4 + +target-version = "py310" + +preview = true + +output-format = "concise" + +extend-exclude = ["tools/mail.py"] + +[tool.ruff.lint] +select = [ + "F", + "E4", + "E7", + "E9", + "W", + "UP", + "YTT", + "S", + "BLE", + "B", + "A", + "C4", + "T10", + "DJ", + "EM", + "EXE", + "ISC", + "ICN", + "G", + "PIE", + "PYI", + "Q003", + "Q004", + "RSE", + "RET", + "SLF", + "SLOT", + "SIM", + "TID", + "TC", + "ARG", + "PGH", + "PL", + "TRY", + "FLY", + "PERF", + "FURB", + "LOG", + "RUF" +] +ignore = [ + "W191", + "PLR09", + "PLR1702", + "PLR2004", + "RUF001", + "RUF002", + "RUF003", + "RUF023" +] + +[tool.ruff.format] +quote-style = "preserve" + +indent-style = "tab" From c64a24e870c7ccc6ab907047764a228d758b6ddc Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:17:03 -0700 Subject: [PATCH 03/37] Fixed UP031 (printf-string-formatting): Use format specifiers instead of percent format --- management/cli.py | 2 +- management/daemon.py | 2 +- management/dns_update.py | 34 +++++++++++++++++----------------- management/mail_log.py | 10 +++++----- management/mailconfig.py | 32 ++++++++++++++++---------------- management/ssl_certificates.py | 13 ++++++------- management/status_checks.py | 26 +++++++++++++------------- management/web_update.py | 10 +++++----- setup/migrate.py | 2 +- tests/fail2ban.py | 2 +- tests/test_dns.py | 6 +++--- tests/test_mail.py | 4 ++-- tools/readable_bash.py | 12 +++++------- 13 files changed, 76 insertions(+), 79 deletions(-) diff --git a/management/cli.py b/management/cli.py index d902b893..f8089607 100755 --- a/management/cli.py +++ b/management/cli.py @@ -120,7 +120,7 @@ elif sys.argv[1] == "user" and sys.argv[2] == "admins": elif sys.argv[1] == "user" and sys.argv[2] == "quota" and len(sys.argv) == 4: # Get a user's quota - print(mgmt("/mail/users/quota?text=1&email=%s" % sys.argv[3])) + print(mgmt("/mail/users/quota?text=1&email={}".format(sys.argv[3]))) elif sys.argv[1] == "user" and sys.argv[2] == "quota" and len(sys.argv) == 5: # Set a user's quota diff --git a/management/daemon.py b/management/daemon.py index 2ad8c480..d1140b5a 100755 --- a/management/daemon.py +++ b/management/daemon.py @@ -743,7 +743,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': '/{}/'.format(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 diff --git a/management/dns_update.py b/management/dns_update.py index 186e14a5..8c931871 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -187,7 +187,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 +254,16 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) # 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 = "Required. May have a different value. Sets the IP address that {} resolves to for web hosting and other services besides mail. The A record must be present but its value does not affect mail delivery.".format(domain) 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 = "Optional. Sets the IP address that {} resolves to so that the box can provide a redirect to the parent domain.".format(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'), "Optional. Sets the IPv6 address that {} resolves to, e.g. for web hosting. (It is not necessary for receiving mail on this domain.)".format(domain)), ] for qname, rtype, value, explanation in defaults: if value is None or value.strip() == "": continue # skip IPV6 if not set @@ -281,13 +281,13 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) 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"]), "Required. Specifies the hostname (and priority) of the machine that handles @{} mail.".format(domain))) # 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', "Recommended. Specifies that only the box is permitted to send @{} mail.".format(domain))) # Append the DKIM TXT record to the zone as generated by OpenDKIM. # Skip if the user has set a DKIM record already. @@ -296,12 +296,12 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) 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, "Recommended. Provides a way for recipients to verify that this machine sent @{} mail.".format(domain))) # 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;', "Recommended. Specifies that mail that does not originate from the box but claims to be from @{} or which does not have a valid DKIM signature is suspect and should be quarantined by the recipient's mail system.".format(domain))) if domain_properties[domain]["user"]: # Add CardDAV/CalDAV SRV records on the non-primary hostname that points to the primary hostname @@ -364,9 +364,9 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) # 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', "Recommended. Prevents use of this domain name for outbound mail by specifying that no servers are valid sources for mail from @{}. 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).".format(d))) 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;', "Recommended. Prevents use of this domain name for outbound mail by specifying that the SPF rule should be honoured for mail from @{}.".format(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 +592,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("{} is not a domain name that corresponds to a zone.".format(zone)) nsd_zonefile = "/etc/nsd/zones/" + fn with open(nsd_zonefile, encoding="utf-8") as f: @@ -617,8 +617,8 @@ zone: # 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 += "\n\tnotify: {} NOKEY".format(ipaddr) + nsdconf += "\n\tprovide-xfr: {} NOKEY\n".format(ipaddr) # Check if the file is changing. If it isn't changing, # return False to flag that no change was made. @@ -898,7 +898,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("{} is not a domain name or a subdomain of a domain name managed by this box.".format(qname)) # validate rtype rtype = rtype.upper() @@ -928,7 +928,7 @@ def set_custom_dns_record(qname, rtype, value, action, env): # anything goes pass else: - raise ValueError("Unknown record type '%s'." % rtype) + raise ValueError("Unknown record type '{}'.".format(rtype)) # load existing config config = list(get_custom_dns_config(env)) @@ -1039,7 +1039,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("Could not resolve the IP address of {}.".format(item)) else: # Validate IP address. try: @@ -1048,7 +1048,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("'{}' is not an IPv4 or IPv6 address or subnet.".format(item[4:])) # Set. set_custom_dns_record("_secondary_nameserver", "A", " ".join(hostnames), "set", env) diff --git a/management/mail_log.py b/management/mail_log.py index 793fec09..8192ca3f 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -302,7 +302,7 @@ def scan_mail_log(env): for date, sender, message in user_data["blocked"]: if len(sender) > 64: sender = sender[:32] + "…" + sender[-32:] - user_rejects.extend((f'{date} - {sender} ', ' %s' % message)) + user_rejects.extend((f'{date} - {sender} ', ' {}'.format(message))) rejects.append(user_rejects) print_user_table( @@ -608,7 +608,7 @@ def valid_date(string): try: date = dateutil.parser.parse(string) except ValueError: - raise argparse.ArgumentTypeError("Unrecognized date and/or time '%s'" % string) + raise argparse.ArgumentTypeError("Unrecognized date and/or time '{}'".format(string)) return date @@ -670,7 +670,7 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None col_str = f"{d[row]!s:<20}" col_left[col] = True else: - temp = "{:>%s}" % max(5, len(l) + 1, len(str(d[row])) + 1) + temp = "{{:>{}}}".format(max(5, len(l) + 1, len(str(d[row])) + 1)) col_str = temp.format(str(d[row])) col_widths[col] = max(col_widths[col], len(col_str)) line += col_str @@ -707,10 +707,10 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None if sub_data is not None: for l, d in sub_data: if d[row]: - lines.extend(('┬', '│ %s' % l, '├─%s─' % (len(l) * '─'), '│')) + lines.extend(('┬', '│ {}'.format(l), '├─%s─' % (len(l) * '─'), '│')) max_len = 0 for v in list(d[row]): - lines.append("│ %s" % v) + lines.append("│ {}".format(v)) max_len = max(max_len, len(v)) lines.append("└" + (max_len + 1) * "─") diff --git a/management/mailconfig.py b/management/mailconfig.py index 68adfba0..f8725e1f 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -108,9 +108,9 @@ def sizeof_fmt(num): for unit in ['','K','M','G','T']: if abs(num) < 1024.0: if abs(num) > 99: - return "%3.0f%s" % (num, unit) + return "{:3.0f}{}".format(num, unit) else: - return "%2.1f%s" % (num, unit) + return "{:2.1f}{}".format(num, unit) num /= 1024.0 @@ -148,7 +148,7 @@ def get_mail_users_ex(env, with_archived=False): box_quota = 0 percent = '' try: - dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/%s/%s/maildirsize' % (domain, user)) + dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/{}/{}/maildirsize'.format(domain, user)) with open(dirsize_file, 'r') as f: box_quota = int(f.readline().split('S')[0]) for line in f.readlines(): @@ -174,7 +174,7 @@ def get_mail_users_ex(env, with_archived=False): "quota": quota, "box_quota": box_quota, "box_size": sizeof_fmt(box_size) if box_size != '?' else box_size, - "percent": '%3.0f%%' % percent if type(percent) != str else percent, + "percent": '{:3.0f}%'.format(percent) if type(percent) != str else percent, "status": "active", } users.append(user) @@ -378,7 +378,7 @@ def set_mail_password(email, pw, env): conn, c = open_database(env, with_connection=True) c.execute("UPDATE users SET password=? WHERE email=?", (pw, email)) if c.rowcount != 1: - return ("That's not a user (%s)." % email, 400) + return ("That's not a user ({}).".format(email), 400) conn.commit() return "OK" @@ -394,7 +394,7 @@ def get_mail_quota(email, env): c.execute("SELECT quota FROM users WHERE email=?", (email,)) rows = c.fetchall() if len(rows) != 1: - return ("That's not a user (%s)." % email, 400) + return ("That's not a user ({}).".format(email), 400) return rows[0][0] @@ -407,7 +407,7 @@ def set_mail_quota(email, quota, env): conn, c = open_database(env, with_connection=True) c.execute("UPDATE users SET quota=? WHERE email=?", (quota, email)) if c.rowcount != 1: - return ("That's not a user (%s)." % email, 400) + return ("That's not a user ({}).".format(email), 400) conn.commit() dovecot_quota_recalc(email) @@ -446,7 +446,7 @@ def get_mail_password(email, env): c.execute('SELECT password FROM users WHERE email=?', (email,)) rows = c.fetchall() if len(rows) != 1: - raise ValueError("That's not a user (%s)." % email) + raise ValueError("That's not a user ({}).".format(email)) return rows[0][0] def remove_mail_user(email, env): @@ -454,7 +454,7 @@ def remove_mail_user(email, env): conn, c = open_database(env, with_connection=True) c.execute("DELETE FROM users WHERE email=?", (email,)) if c.rowcount != 1: - return ("That's not a user (%s)." % email, 400) + return ("That's not a user ({}).".format(email), 400) conn.commit() # Update things in case any domains are removed. @@ -470,12 +470,12 @@ def get_mail_user_privileges(email, env, empty_on_error=False): rows = c.fetchall() if len(rows) != 1: if empty_on_error: return [] - return ("That's not a user (%s)." % email, 400) + return ("That's not a user ({}).".format(email), 400) return parse_privs(rows[0][0]) def validate_privilege(priv): if "\n" in priv or priv.strip() == "": - return ("That's not a valid privilege (%s)." % priv, 400) + return ("That's not a valid privilege ({}).".format(priv), 400) return None def add_remove_mail_user_privilege(email, priv, action, env): @@ -518,7 +518,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist if address == "": return ("No email address provided.", 400) if not validate_email(address, mode='alias'): - return ("Invalid email address (%s)." % address, 400) + return ("Invalid email address ({}).".format(address), 400) # validate forwards_to validated_forwards_to = [] @@ -547,7 +547,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist # Strip any +tag from email alias and check privileges privileged_email = re.sub(r"(?=\+)[^@]*(?=@)",'',email) if not validate_email(email): - return ("Invalid receiver email address (%s)." % email, 400) + return ("Invalid receiver email address ({}).".format(email), 400) if is_dcv_source and not is_dcv_address(email) and "admin" not in get_mail_user_privileges(privileged_email, env, empty_on_error=True): # Make domain control validation hijacking a little harder to mess up by # requiring aliases for email addresses typically used in DCV to forward @@ -567,7 +567,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist login = login.strip() if login == "": continue if login not in valid_logins: - return ("Invalid permitted sender: %s is not a user on this system." % login, 400) + return ("Invalid permitted sender: {} is not a user on this system.".format(login), 400) validated_permitted_senders.append(login) # Make sure the alias has either a forwards_to or a permitted_sender. @@ -586,7 +586,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist return_status = "alias added" except sqlite3.IntegrityError: if not update_if_exists: - return ("Alias already exists (%s)." % address, 400) + return ("Alias already exists ({}).".format(address), 400) else: c.execute("UPDATE aliases SET destination = ?, permitted_senders = ? WHERE source = ?", (forwards_to, permitted_senders, address)) return_status = "alias updated" @@ -606,7 +606,7 @@ def remove_mail_alias(address, env, do_kick=True): conn, c = open_database(env, with_connection=True) c.execute("DELETE FROM aliases WHERE source=?", (address,)) if c.rowcount != 1: - return ("That's not an alias (%s)." % address, 400) + return ("That's not an alias ({}).".format(address), 400) conn.commit() if do_kick: diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index 8c1b841e..af262ba4 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -518,7 +518,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring cert = load_pem(ssl_cert_chain[0]) if not isinstance(cert, Certificate): raise ValueError("This is not a certificate file.") except ValueError as e: - return ("There is a problem with the certificate file: %s" % str(e), None) + return ("There is a problem with the certificate file: {}".format(str(e)), None) # First check that the domain name is one of the names allowed by # the certificate. @@ -530,8 +530,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring # should work in normal cases). wildcard_domain = re.sub(r"^[^\.]+", "*", domain) if domain not in certificate_names and wildcard_domain not in certificate_names: - return ("The certificate is for the wrong domain name. It is for %s." - % ", ".join(sorted(certificate_names)), None) + return ("The certificate is for the wrong domain name. It is for {}.".format(", ".join(sorted(certificate_names))), None) # Second, check that the certificate matches the private key. if ssl_private_key is not None: @@ -544,10 +543,10 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring if (not isinstance(priv_key, rsa.RSAPrivateKey) and not isinstance(priv_key, dsa.DSAPrivateKey) and not isinstance(priv_key, ec.EllipticCurvePrivateKey)): - return ("The private key file %s is not a private key file." % ssl_private_key, None) + return ("The private key file {} is not a private key file.".format(ssl_private_key), None) if priv_key.public_key().public_numbers() != cert.public_key().public_numbers(): - return ("The certificate does not correspond to the private key at %s." % ssl_private_key, None) + return ("The certificate does not correspond to the private key at {}.".format(ssl_private_key), None) # We could also use the openssl command line tool to get the modulus # listed in each file. The output of each command below looks like "Modulus=XXXXX". @@ -593,7 +592,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring elif retcode != 0: if "unable to get local issuer certificate" in verifyoutput: - return ("The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. (%s)" % verifyoutput, None) + return ("The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. ({})".format(verifyoutput), None) # There is some unknown problem. Return the `openssl verify` raw output. return ("There is a problem with the certificate.", verifyoutput.strip()) @@ -610,7 +609,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring expiry_info = "The certificate expires in %d days on %s." % (ndays, cert_expiration_date.date().isoformat()) else: # We'll renew it with Lets Encrypt. - expiry_info = "The certificate expires on %s." % cert_expiration_date.date().isoformat() + expiry_info = "The certificate expires on {}.".format(cert_expiration_date.date().isoformat()) if warn_if_expiring_soon and ndays <= warn_if_expiring_soon: # Warn on day 10 to give 4 days for us to automatically renew the diff --git a/management/status_checks.py b/management/status_checks.py index 68755cb7..5ea9c18e 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -248,7 +248,7 @@ def check_free_disk_space(rounded_values, env, output): def check_free_memory(rounded_values, env, output): # Check free memory. percent_free = 100 - psutil.virtual_memory().percent - memory_msg = "System memory is %s%% free." % str(round(percent_free)) + memory_msg = "System memory is {}% free.".format(str(round(percent_free))) if percent_free >= 20: if rounded_values: memory_msg = "System free memory is at least 20%." output.print_ok(memory_msg) @@ -478,7 +478,7 @@ def check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles): tlsa25 = query_dns(tlsa_qname, "TLSA", nxdomain=None) tlsa25_expected = build_tlsa_record(env) if tlsa25 == tlsa25_expected: - output.print_ok("""The DANE TLSA record for incoming mail is correct (%s).""" % tlsa_qname,) + output.print_ok("""The DANE TLSA record for incoming mail is correct ({}).""".format(tlsa_qname),) elif tlsa25 is None: if has_dnssec: # Omit a warning about it not being set if DNSSEC isn't enabled, @@ -497,9 +497,9 @@ def check_alias_exists(alias_name, alias, env, output): if mail_aliases[alias]: output.print_ok(f"{alias_name} exists as a mail alias. [{alias} ↦ {mail_aliases[alias]}]") else: - output.print_error("""You must set the destination of the mail alias for %s to direct email to you or another administrator.""" % alias) + output.print_error("""You must set the destination of the mail alias for {} to direct email to you or another administrator.""".format(alias)) else: - output.print_error("""You must add a mail alias for %s which directs email to you or another administrator.""" % alias) + output.print_error("""You must add a mail alias for {} which directs email to you or another administrator.""".format(alias)) def check_dns_zone(domain, env, output, dns_zonefiles): # If a DS record is set at the registrar, check DNSSEC first because it will affect the NS query. @@ -527,7 +527,7 @@ def check_dns_zone(domain, env, output, dns_zonefiles): probably_external_dns = False if existing_ns.lower() == correct_ns.lower(): - output.print_ok("Nameservers are set correctly at registrar. [%s]" % correct_ns) + output.print_ok("Nameservers are set correctly at registrar. [{}]".format(correct_ns)) elif ip == correct_ip: # The domain resolves correctly, so maybe the user is using External DNS. output.print_warning(f"""The nameservers set on this domain at your domain name registrar should be {correct_ns}. They are currently {existing_ns}. @@ -546,7 +546,7 @@ def check_dns_zone(domain, env, output, dns_zonefiles): # We must first resolve the nameserver to an IP address so we can query it. ns_ips = query_dns(ns, "A") if not ns_ips or ns_ips in {'[Not Set]', '[timeout]'}: - output.print_error("Secondary nameserver %s is not valid (it doesn't resolve to an IP address)." % ns) + output.print_error("Secondary nameserver {} is not valid (it doesn't resolve to an IP address).".format(ns)) continue # Choose the first IP if nameserver returns multiple ns_ip = ns_ips.split('; ')[0] @@ -587,7 +587,7 @@ def check_dns_zone_suggestions(domain, env, output, dns_zonefiles, domains_with_ if domain in domains_with_a_records: output.print_warning("""Web has been disabled for this domain because you have set a custom DNS record.""") if "www." + domain in domains_with_a_records: - output.print_warning("""A redirect from 'www.%s' has been disabled for this domain because you have set a custom DNS record on the www subdomain.""" % domain) + output.print_warning("""A redirect from 'www.{}' has been disabled for this domain because you have set a custom DNS record on the www subdomain.""".format(domain)) # Since DNSSEC is optional, if a DS record is NOT set at the registrar suggest it. # (If it was set, we did the check earlier.) @@ -616,7 +616,7 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): # Some registrars may want the public key so they can compute the digest. The DS # record that we suggest using is for the KSK (and that's how the DS records were generated). # We'll also give the nice name for the key algorithm. - dnssec_keys = load_env_vars_from_file(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/%s.conf' % alg_name_map[ds_alg])) + dnssec_keys = load_env_vars_from_file(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/{}.conf'.format(alg_name_map[ds_alg]))) with open(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/' + dnssec_keys['KSK'] + '.key'), encoding="utf-8") as f: dnsssec_pubkey = f.read().split("\t")[3].split(" ")[3] @@ -662,7 +662,7 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): for this domain is valid.""") else: if is_checking_primary: - output.print_error("""The DNSSEC 'DS' record for %s is incorrect. See further details below.""" % domain) + output.print_error("""The DNSSEC 'DS' record for {} is incorrect. See further details below.""".format(domain)) return output.print_error("""This domain's DNSSEC DS record is incorrect. The chain of trust is broken between the public DNS system and this machine's DNS server. It may take several hours for public DNS to update after a change. If you did not recently @@ -774,7 +774,7 @@ def check_mail_domain(domain, env, output): elif dbl == "[Not Set]": output.print_warning(f"Could not connect to dbl.spamhaus.org. Could not determine whether the domain {domain} is blacklisted. Please try again later.") elif dbl == "127.255.255.252": - output.print_warning("Incorrect spamhaus query: %s. Could not determine whether the domain %s is blacklisted." % (domain+'.dbl.spamhaus.org', domain)) + output.print_warning("Incorrect spamhaus query: {}. Could not determine whether the domain {} is blacklisted.".format(domain+'.dbl.spamhaus.org', domain)) elif dbl == "127.255.255.254": output.print_warning("Mail-in-a-Box is configured to use a public DNS server. This is not supported by spamhaus. Could not determine whether the domain {} is blacklisted.".format(domain)) elif dbl == "127.255.255.255": @@ -960,14 +960,14 @@ def check_miab_version(env, output): this_ver = "Unknown" if config.get("privacy", True): - output.print_warning("You are running version Mail-in-a-Box %s. Mail-in-a-Box version check disabled by privacy setting." % this_ver) + output.print_warning("You are running version Mail-in-a-Box {}. Mail-in-a-Box version check disabled by privacy setting.".format(this_ver)) else: latest_ver = get_latest_miab_version() if this_ver == latest_ver: - output.print_ok("Mail-in-a-Box is up to date. You are running version %s." % this_ver) + output.print_ok("Mail-in-a-Box is up to date. You are running version {}.".format(this_ver)) elif latest_ver is None: - output.print_error("Latest Mail-in-a-Box version could not be determined. You are running version %s." % this_ver) + output.print_error("Latest Mail-in-a-Box version could not be determined. You are running version {}.".format(this_ver)) else: output.print_error(f"A new version of Mail-in-a-Box is available. You are running version {this_ver}. The latest version is {latest_ver}. For upgrade instructions, see https://mailinabox.email. ") diff --git a/management/web_update.py b/management/web_update.py index e97bc60d..265d7b9e 100644 --- a/management/web_update.py +++ b/management/web_update.py @@ -180,8 +180,8 @@ def make_domain_config(domain, templates, ssl_certificates, env): web_sockets = True url = re.sub("#(.*)$", "", url) - nginx_conf_extra += "\tlocation %s {" % path - nginx_conf_extra += "\n\t\tproxy_pass %s;" % url + nginx_conf_extra += "\tlocation {} {{".format(path) + nginx_conf_extra += "\n\t\tproxy_pass {};".format(url) if proxy_redirect_off: nginx_conf_extra += "\n\t\tproxy_redirect off;" if pass_http_host_header: @@ -198,8 +198,8 @@ def make_domain_config(domain, templates, ssl_certificates, env): nginx_conf_extra += "\n\t\tproxy_set_header X-Real-IP $remote_addr;" nginx_conf_extra += "\n\t}\n" for path, alias in yaml.get("aliases", {}).items(): - nginx_conf_extra += "\tlocation %s {" % path - nginx_conf_extra += "\n\t\talias %s;" % alias + nginx_conf_extra += "\tlocation {} {{".format(path) + nginx_conf_extra += "\n\t\talias {};".format(alias) nginx_conf_extra += "\n\t}\n" for path, url in yaml.get("redirects", {}).items(): nginx_conf_extra += f"\trewrite {path} {url} permanent;\n" @@ -216,7 +216,7 @@ def make_domain_config(domain, templates, ssl_certificates, env): # Add in any user customizations in the includes/ folder. nginx_conf_custom_include = os.path.join(env["STORAGE_ROOT"], "www", safe_domain_name(domain) + ".conf") if os.path.exists(nginx_conf_custom_include): - nginx_conf_extra += "\tinclude %s;\n" % (nginx_conf_custom_include) + nginx_conf_extra += "\tinclude {};\n".format(nginx_conf_custom_include) # PUT IT ALL TOGETHER # Combine the pieces. Iteratively place each template into the "# ADDITIONAL DIRECTIVES HERE" placeholder diff --git a/setup/migrate.py b/setup/migrate.py index 4c161e45..dbfc54b5 100755 --- a/setup/migrate.py +++ b/setup/migrate.py @@ -164,7 +164,7 @@ def migration_12(env): try: table = table[0] c = conn.cursor() - dropcmd = "DROP TABLE %s" % table + dropcmd = "DROP TABLE {}".format(table) c.execute(dropcmd) except: print("Failed to drop table", table) diff --git a/tests/fail2ban.py b/tests/fail2ban.py index dbab874a..abddcc41 100644 --- a/tests/fail2ban.py +++ b/tests/fail2ban.py @@ -142,7 +142,7 @@ def http_test(url, expected_status, postdata=None, qsargs=None, auth=None): # return response status code if r.status_code != expected_status: r.raise_for_status() # anything but 200 - raise OSError("Got unexpected status code %s." % r.status_code) + raise OSError("Got unexpected status code {}.".format(r.status_code)) # define how to run a test diff --git a/tests/test_dns.py b/tests/test_dns.py index 29b681cc..6f84d5d0 100755 --- a/tests/test_dns.py +++ b/tests/test_dns.py @@ -51,7 +51,7 @@ def test2(tests, server, description): response = dns.resolver.resolve(qname, rtype) except dns.resolver.NoNameservers: # host did not have an answer for this query - print("Could not connect to %s for DNS query." % server) + print("Could not connect to {} for DNS query.".format(server)) sys.exit(1) except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): # host did not have an answer for this query; not sure what the @@ -79,7 +79,7 @@ def test2(tests, server, description): # Test the response from the machine itself. if not test(ipaddr, "Mail-in-a-Box"): print () - print ("Please run the Mail-in-a-Box setup script on %s again." % hostname) + print ("Please run the Mail-in-a-Box setup script on {} again.".format(hostname)) sys.exit(1) else: print ("The Mail-in-a-Box provided correct DNS answers.") @@ -89,7 +89,7 @@ else: # to see if the machine is hooked up to recursive DNS properly. if not test("8.8.8.8", "Google Public DNS"): print () - print ("Check that the nameserver settings for %s are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box." % hostname) + print ("Check that the nameserver settings for {} are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box.".format(hostname)) sys.exit(1) else: print ("Your domain registrar or DNS host appears to be configured correctly as well. Public DNS provides the same answers.") diff --git a/tests/test_mail.py b/tests/test_mail.py index 64ed3679..9cc7bb03 100755 --- a/tests/test_mail.py +++ b/tests/test_mail.py @@ -46,7 +46,7 @@ reverse_ip = dns.reversename.from_address(ipaddr) # e.g. "1.0.0.127.in-addr.arpa try: reverse_dns = dns.resolver.resolve(reverse_ip, 'PTR')[0].target.to_text(omit_final_dot=True) # => hostname except dns.resolver.NXDOMAIN: - print("Reverse DNS lookup failed for %s. SMTP EHLO name check skipped." % ipaddr) + print("Reverse DNS lookup failed for {}. SMTP EHLO name check skipped.".format(ipaddr)) reverse_dns = None if reverse_dns is not None: server.ehlo_or_helo_if_needed() # must send EHLO before getting the server's EHLO name @@ -54,7 +54,7 @@ if reverse_dns is not None: if helo_name != reverse_dns: print("The server's EHLO name does not match its reverse hostname. Check DNS settings.") else: - print("SMTP EHLO name (%s) is OK." % helo_name) + print("SMTP EHLO name ({}) is OK.".format(helo_name)) # Login and send a test email. server.login(emailaddress, pw) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index eb439397..ac4b6448 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -192,8 +192,7 @@ class CatEOF(Grammar): def value(self): content = self[9].string content = re.sub(r"\\([$])", r"\1", content) # un-escape bash-escaped characters - return "
%s (%s)
%s
\n" \ - % (self[4].string, + return "
{} ({})
{}
\n".format(self[4].string, "overwrite" if ">>" not in self[2].string else "append to", cgi.escape(content)) @@ -230,7 +229,7 @@ class EditConf(Grammar): for opt in re.split("\s+", self[4].string): k, v = opt.split("=", 1) v = re.sub(r"\n+", "", fixup_tokens(v)) # not sure why newlines are getting doubled - options.append("%s%s%s" % (k, eq, v)) + options.append("{}{}{}".format(k, eq, v)) return "
" + self[1].string + " (change settings)
" + "\n".join(cgi.escape(s) for s in options) + "
\n" class CaptureOutput(Grammar): @@ -248,7 +247,7 @@ class SedReplace(Grammar): class EchoPipe(Grammar): grammar = OPTIONAL(SPACE), L("echo "), REST_OF_LINE, L(' | '), REST_OF_LINE, EOL def value(self): - text = " ".join("\"%s\"" % s for s in self[2].string.split(" ")) + text = " ".join("\"{}\"".format(s) for s in self[2].string.split(" ")) return "
echo " + recode_bash(text) + " \
| " + recode_bash(self[4].string) + "
\n" def shell_line(bash): @@ -414,8 +413,7 @@ class BashScript(Grammar): parser = BashScript.parser() result = parser.parse_string(string) - v = "
view the bash source for the following section at %s
\n" \ - % ("https://github.com/mail-in-a-box/mailinabox/tree/master/" + fn, fn) + v = "
view the bash source for the following section at {}
\n".format("https://github.com/mail-in-a-box/mailinabox/tree/master/" + fn, fn) mode = 0 for item in result.value(): @@ -429,7 +427,7 @@ class BashScript(Grammar): mode = 0 clz = "contd" if mode == 0: - v += "
\n" % clz + v += "
\n".format(clz) v += "
\n" v += item mode = 1 From 68fd3dc535b95ff8e9511d7495ddc8d462c5336d Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:17:27 -0700 Subject: [PATCH 04/37] Fixed FURB118 (reimplemented-operator) --- management/backup.py | 3 ++- management/mailconfig.py | 3 ++- 2 files changed, 4 insertions(+), 2 deletions(-) diff --git a/management/backup.py b/management/backup.py index 311db3c3..900beff8 100755 --- a/management/backup.py +++ b/management/backup.py @@ -14,6 +14,7 @@ import rtyaml 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 backup_status(env): # 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 diff --git a/management/mailconfig.py b/management/mailconfig.py index f8725e1f..9e883121 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -15,6 +15,7 @@ import subprocess import utils from email_validator import validate_email as validate_email_, EmailNotValidError import idna +import operator def validate_email(email, mode=None): # Checks that an email address is syntactically valid. Returns True/False. @@ -285,7 +286,7 @@ def get_mail_aliases_ex(env): # Sort aliases within each domain first by required-ness then lexicographically by address. for domain in domains: - domain["aliases"].sort(key = lambda alias : (alias["auto"], alias["address"])) + domain["aliases"].sort(key = operator.itemgetter("auto", "address")) return domains def get_domain(emailaddr, as_unicode=True): From 3a1280d2921ff4ef22e017ab2d5dbda4be2bbe17 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:19:08 -0700 Subject: [PATCH 05/37] Fixed PLW0120 (useless-else-on-loop): `else` clause on loop without a `break` statement; remove the `else` and dedent its contents --- management/backup.py | 6 ++---- 1 file changed, 2 insertions(+), 4 deletions(-) diff --git a/management/backup.py b/management/backup.py index 900beff8..2b90efd0 100755 --- a/management/backup.py +++ b/management/backup.py @@ -178,10 +178,8 @@ 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. + return True def get_passphrase(env): # Get the encryption passphrase. secret_key.txt is 2048 random From 3008dfa28fa5d20f29cd3f2c2d8973226fc8a450 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:19:36 -0700 Subject: [PATCH 06/37] Fixed EM101 (raw-string-in-exception): Exception must not use a string literal, assign to variable first --- management/mailconfig.py | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/management/mailconfig.py b/management/mailconfig.py index 9e883121..98d8f005 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -430,11 +430,14 @@ def validate_quota(quota): quota = quota.strip().upper() if quota == "": - raise ValueError("No quota provided.") + msg = "No quota provided." + raise ValueError(msg) if re.search(r"[\s,.]", quota): - raise ValueError("Quotas cannot contain spaces, commas, or decimal points.") + msg = "Quotas cannot contain spaces, commas, or decimal points." + raise ValueError(msg) if not re.match(r'^[\d]+[GM]?$', quota): - raise ValueError("Invalid quota.") + msg = "Invalid quota." + raise ValueError(msg) return quota From dbabd69218fc13f31b429bb899e30543ea42e24b Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:19:55 -0700 Subject: [PATCH 07/37] Fixed RET505 (superfluous-else-return) --- management/backup.py | 31 +++++++++++---------- management/daemon.py | 42 +++++++++++++---------------- management/dns_update.py | 3 +-- management/mail_log.py | 5 ++-- management/mailconfig.py | 17 +++++------- management/ssl_certificates.py | 49 ++++++++++++++++------------------ management/status_checks.py | 6 ++--- management/utils.py | 3 +-- management/web_update.py | 5 ++-- 9 files changed, 72 insertions(+), 89 deletions(-) diff --git a/management/backup.py b/management/backup.py index 2b90efd0..08902a57 100755 --- a/management/backup.py +++ b/management/backup.py @@ -235,7 +235,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. @@ -446,7 +446,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}' @@ -484,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 diff --git a/management/daemon.py b/management/daemon.py index d1140b5a..6658af09 100755 --- a/management/daemon.py +++ b/management/daemon.py @@ -93,12 +93,11 @@ def authorized_personnel_only(viewfunc): 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 @@ -148,13 +147,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 = { @@ -186,8 +184,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 @@ -257,8 +254,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 @@ -378,7 +374,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 @@ -536,8 +532,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 @@ -621,8 +617,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 @@ -631,8 +626,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') diff --git a/management/dns_update.py b/management/dns_update.py index 8c931871..1cc8a0da 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -124,8 +124,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" ######################################################################## diff --git a/management/mail_log.py b/management/mail_log.py index 8192ca3f..7d1cede2 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -355,7 +355,7 @@ def scan_mail_log_line(line, collector): if date > END_DATE: # Don't process, and halt return False - elif date < START_DATE: + if date < START_DATE: # Don't process, but continue return True @@ -634,8 +634,7 @@ def print_time_table(labels, data, do_print=True): if do_print: print("\n".join(lines)) return None - else: - return lines + return lines def print_user_table(users, data=None, sub_data=None, activity=None, latest=None, earliest=None, diff --git a/management/mailconfig.py b/management/mailconfig.py index 98d8f005..7caee5f7 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -95,8 +95,7 @@ def open_database(env, with_connection=False): conn = sqlite3.connect(env["STORAGE_ROOT"] + "/mail/users.sqlite") if not with_connection: return conn.cursor() - else: - return conn, conn.cursor() + return conn, conn.cursor() def get_mail_users(env): # Returns a flat, sorted list of all user accounts. @@ -110,8 +109,7 @@ def sizeof_fmt(num): if abs(num) < 1024.0: if abs(num) > 99: return "{:3.0f}{}".format(num, unit) - else: - return "{:2.1f}{}".format(num, unit) + return "{:2.1f}{}".format(num, unit) num /= 1024.0 @@ -317,11 +315,11 @@ def add_mail_user(email, pw, privs, quota, env): # validate email if email.strip() == "": return ("No email address provided.", 400) - elif not validate_email(email): + if not validate_email(email): return ("Invalid email address.", 400) - elif not validate_email(email, mode='user'): + if not validate_email(email, mode='user'): return ("User account email addresses may only use the lowercase ASCII letters a-z, the digits 0-9, underscore (_), hyphen (-), and period (.).", 400) - elif is_dcv_address(email) and len(get_mail_users(env)) > 0: + if is_dcv_address(email) and len(get_mail_users(env)) > 0: # Make domain control validation hijacking a little harder to mess up by preventing the usual # addresses used for DCV from being user accounts. Except let it be the first account because # during box setup the user won't know the rules. @@ -591,9 +589,8 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist except sqlite3.IntegrityError: if not update_if_exists: return ("Alias already exists ({}).".format(address), 400) - else: - c.execute("UPDATE aliases SET destination = ?, permitted_senders = ? WHERE source = ?", (forwards_to, permitted_senders, address)) - return_status = "alias updated" + c.execute("UPDATE aliases SET destination = ?, permitted_senders = ? WHERE source = ?", (forwards_to, permitted_senders, address)) + return_status = "alias updated" conn.commit() diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index af262ba4..effb6ace 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -160,14 +160,13 @@ def get_domain_ssl_files(domain, ssl_certificates, env, allow_missing_cert=False wildcard_domain = re.sub(r"^[^\.]+", "*", domain) if domain in ssl_certificates: return ssl_certificates[domain] - elif wildcard_domain in ssl_certificates: + if wildcard_domain in ssl_certificates: return ssl_certificates[wildcard_domain] - elif not allow_missing_cert: + if not allow_missing_cert: # No valid certificate is available for this domain! Return default files. return system_certificate - else: - # No valid certificate is available for this domain. - return None + # No valid certificate is available for this domain. + return None # PROVISIONING CERTIFICATES FROM LETSENCRYPT @@ -590,34 +589,33 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring # Certificate is self-signed. Probably we detected this above. return ("SELF-SIGNED", None) - elif retcode != 0: + if retcode != 0: if "unable to get local issuer certificate" in verifyoutput: return ("The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. ({})".format(verifyoutput), None) # There is some unknown problem. Return the `openssl verify` raw output. return ("There is a problem with the certificate.", verifyoutput.strip()) + # `openssl verify` returned a zero exit status so the cert is currently + # good. + + # But is it expiring soon? + cert_expiration_date = cert.not_valid_after + ndays = (cert_expiration_date-now).days + if not rounded_time or ndays <= 10: + # Yikes better renew soon! + expiry_info = "The certificate expires in %d days on %s." % (ndays, cert_expiration_date.date().isoformat()) else: - # `openssl verify` returned a zero exit status so the cert is currently - # good. + # We'll renew it with Lets Encrypt. + expiry_info = "The certificate expires on {}.".format(cert_expiration_date.date().isoformat()) - # But is it expiring soon? - cert_expiration_date = cert.not_valid_after - ndays = (cert_expiration_date-now).days - if not rounded_time or ndays <= 10: - # Yikes better renew soon! - expiry_info = "The certificate expires in %d days on %s." % (ndays, cert_expiration_date.date().isoformat()) - else: - # We'll renew it with Lets Encrypt. - expiry_info = "The certificate expires on {}.".format(cert_expiration_date.date().isoformat()) + if warn_if_expiring_soon and ndays <= warn_if_expiring_soon: + # Warn on day 10 to give 4 days for us to automatically renew the + # certificate, which occurs on day 14. + return ("The certificate is expiring soon: " + expiry_info, None) - if warn_if_expiring_soon and ndays <= warn_if_expiring_soon: - # Warn on day 10 to give 4 days for us to automatically renew the - # certificate, which occurs on day 14. - return ("The certificate is expiring soon: " + expiry_info, None) - - # Return the special OK code. - return ("OK", expiry_info) + # Return the special OK code. + return ("OK", expiry_info) def load_cert_chain(pemfile): # A certificate .pem file may contain a chain of certificates. @@ -671,8 +669,7 @@ def get_certificate_domains(cert): def idna_decode_dns_name(dns_name): if dns_name.startswith("*."): return "*." + idna.encode(dns_name[2:]).decode('ascii') - else: - return idna.encode(dns_name).decode('ascii') + return idna.encode(dns_name).decode('ascii') try: sans = cert.extensions.get_extension_for_oid(OID_SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(DNSName) diff --git a/management/status_checks.py b/management/status_checks.py index 5ea9c18e..a282be88 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -653,11 +653,11 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): if {r[1] for r in matched_ds} == { '13' } and {r[2] for r in matched_ds} <= { '2', '4' }: # all are alg 13 and digest type 2 or 4 output.print_ok("DNSSEC 'DS' record is set correctly at registrar.") return - elif len([r for r in matched_ds if r[1] == '13' and r[2] in { '2', '4' }]) > 0: # some but not all are alg 13 + if len([r for r in matched_ds if r[1] == '13' and r[2] in { '2', '4' }]) > 0: # some but not all are alg 13 output.print_ok("DNSSEC 'DS' record is set correctly at registrar. (Records using algorithm other than ECDSAP256SHA256 and digest types other than SHA-256/384 should be removed.)") return - else: # no record uses alg 13 - output.print_warning("""DNSSEC 'DS' record set at registrar is valid but should be updated to ECDSAP256SHA256 and SHA-256 (see below). + # no record uses alg 13 + output.print_warning("""DNSSEC 'DS' record set at registrar is valid but should be updated to ECDSAP256SHA256 and SHA-256 (see below). IMPORTANT: Do not delete existing DNSSEC 'DS' records for this domain until confirmation that the new DNSSEC 'DS' record for this domain is valid.""") else: diff --git a/management/utils.py b/management/utils.py index 1dbbeb7e..1a5b2195 100644 --- a/management/utils.py +++ b/management/utils.py @@ -135,8 +135,7 @@ def shell(method, cmd_args, env=None, capture_stderr=False, return_bytes=False, if not return_bytes and isinstance(ret, bytes): ret = ret.decode("utf8") if not trap: return ret - else: - return code, ret + return code, ret def create_syslog_handler(): import logging.handlers diff --git a/management/web_update.py b/management/web_update.py index 265d7b9e..708a8dff 100644 --- a/management/web_update.py +++ b/management/web_update.py @@ -256,10 +256,9 @@ def get_web_domains_info(env): cert_status, cert_status_details = check_certificate(domain, tls_cert["certificate"], tls_cert["private-key"]) if cert_status == "OK": return ("success", "Signed & valid. " + cert_status_details) - elif cert_status == "SELF-SIGNED": + if cert_status == "SELF-SIGNED": return ("warning", "Self-signed. Get a signed certificate to stop warnings.") - else: - return ("danger", "Certificate has a problem: " + cert_status) + return ("danger", "Certificate has a problem: " + cert_status) return [ { From ddee3c6bfd161c21021d9c03f1b31e196d11aa4c Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:20:08 -0700 Subject: [PATCH 08/37] Fixed PLR6104 (non-augmented-assignment): Use `+=` to perform an augmented assignment directly --- management/backup.py | 2 +- management/dns_update.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/management/backup.py b/management/backup.py index 08902a57..1887af79 100755 --- a/management/backup.py +++ b/management/backup.py @@ -462,7 +462,7 @@ def list_target_files(config): target_path = target.path if not target_path.endswith('/'): - target_path = target_path + '/' + target_path += '/' if target_path.startswith('/'): target_path = target_path[1:] diff --git a/management/dns_update.py b/management/dns_update.py index 1cc8a0da..ffb97ebe 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -918,7 +918,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." From a83db1aebcf852a1d7ce488f4636d7c1b32dc11a Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:20:19 -0700 Subject: [PATCH 09/37] Fixed FURB188 (slice-to-remove-prefix-or-suffix): Prefer `str.removeprefix()` over conditionally replacing with slice. --- management/backup.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/management/backup.py b/management/backup.py index 1887af79..d8f8c50e 100755 --- a/management/backup.py +++ b/management/backup.py @@ -463,8 +463,7 @@ def list_target_files(config): target_path = target.path if not target_path.endswith('/'): target_path += '/' - if target_path.startswith('/'): - target_path = target_path[1:] + target_path = target_path.removeprefix('/') rsync_command = [ 'rsync', '-e', From d5d4ba0bf124fd57713b59d88a4d7d1bbec28e99 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:21:49 -0700 Subject: [PATCH 10/37] Fixed RUF051 (if-key-in-dict-del): Use `pop` instead of `key in dict` followed by `del dict[key]` --- management/backup.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/management/backup.py b/management/backup.py index d8f8c50e..ecd3ab1e 100755 --- a/management/backup.py +++ b/management/backup.py @@ -602,8 +602,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') From 3fa0819e04b6a433e6f8d696a298ee5530afc034 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:24:37 -0700 Subject: [PATCH 11/37] Fixed G004 (logging-f-string): Logging statement uses f-string --- management/daemon.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/management/daemon.py b/management/daemon.py index 6658af09..67a5d7dc 100755 --- a/management/daemon.py +++ b/management/daemon.py @@ -162,7 +162,7 @@ def login(): "api_key": auth_service.create_session_key(email, env, type='login'), } - app.logger.info(f"New login session created for {email}") + app.logger.info("New login session created for %s", email) # Return. return json_response(resp) @@ -171,7 +171,7 @@ def login(): def logout(): try: email, _ = auth_service.authenticate(request, env, logout=True) - app.logger.info(f"{email} logged out") + app.logger.info("%s logged out", email) except ValueError: pass finally: @@ -771,7 +771,7 @@ def log_failed_login(request): # We need to add a timestamp to the log message, otherwise /dev/log will eat the "duplicate" # message. - app.logger.warning( f"Mail-in-a-Box Management Daemon: Failed login attempt from ip {ip} - timestamp {time.time()}") + app.logger.warning("Mail-in-a-Box Management Daemon: Failed login attempt from ip %s - timestamp %s", ip, time.time()) # APP From e3ef6d726b1459131b646075306afd6541d5ba86 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:25:12 -0700 Subject: [PATCH 12/37] Fixed SIM103 (needless-bool): Return the condition `"admin" in privs` directly --- management/daemon.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/management/daemon.py b/management/daemon.py index 67a5d7dc..7b66f196 100755 --- a/management/daemon.py +++ b/management/daemon.py @@ -688,8 +688,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) From 0635e89b6e8c1a4f5e547f1cd908487d12f429de Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:25:32 -0700 Subject: [PATCH 13/37] Fixed F401 (unused-import): `contextlib` imported but unused --- management/dns_update.py | 1 - 1 file changed, 1 deletion(-) diff --git a/management/dns_update.py b/management/dns_update.py index ffb97ebe..dbb25a4e 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -11,7 +11,6 @@ import dns.resolver 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, From e73771be5ffcf44184f9a893a3f29d853ee832a4 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:27:07 -0700 Subject: [PATCH 14/37] Fixed RET507 (superfluous-else-continue): Unnecessary `elif` after `continue` statement --- management/mail_log.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management/mail_log.py b/management/mail_log.py index 7d1cede2..ad645d90 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -71,7 +71,7 @@ def scan_files(collector): if not os.path.exists(fn): continue - elif fn[-3:] == '.gz': + if fn[-3:] == '.gz': tmp_file = tempfile.NamedTemporaryFile() with gzip.open(fn, 'rb') as f: shutil.copyfileobj(f, tmp_file) From eae0db9df1e6bc35f9f3f65d355ac082e8a91b72 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:27:31 -0700 Subject: [PATCH 15/37] Fixed RUF039 (unraw-re-pattern) --- management/mail_log.py | 6 +++--- management/web_update.py | 4 ++-- setup/migrate.py | 2 +- tests/tls.py | 2 +- tools/readable_bash.py | 2 +- 5 files changed, 8 insertions(+), 8 deletions(-) diff --git a/management/mail_log.py b/management/mail_log.py index ad645d90..5abacc4f 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -391,7 +391,7 @@ def scan_postgrey_line(date, log, collector): """ Scan a postgrey log line and extract interesting data """ m = re.match(r"action=(greylist|pass), reason=(.*?), (?:delay=\d+, )?client_name=(.*), " - "client_address=(.*), sender=(.*), recipient=(.*)", + r"client_address=(.*), sender=(.*), recipient=(.*)", log) if m: @@ -423,7 +423,7 @@ def scan_postfix_smtpd_line(date, log, collector): # Check if the incoming mail was rejected - m = re.match("NOQUEUE: reject: RCPT from .*?: (.*?); from=<(.*?)> to=<(.*?)>", log) + m = re.match(r"NOQUEUE: reject: RCPT from .*?: (.*?); from=<(.*?)> to=<(.*?)>", log) if m: message, sender, user = m.groups() @@ -467,7 +467,7 @@ def scan_postfix_smtpd_line(date, log, collector): def scan_dovecot_login_line(date, log, collector, protocol_name): """ Scan a dovecot login log line and extract interesting data """ - m = re.match("Info: Login: user=<(.*?)>, method=PLAIN, rip=(.*?),", log) + m = re.match(r"Info: Login: user=<(.*?)>, method=PLAIN, rip=(.*?),", log) if m: # TODO: CHECK DIT diff --git a/management/web_update.py b/management/web_update.py index 708a8dff..2d7ec0b2 100644 --- a/management/web_update.py +++ b/management/web_update.py @@ -167,7 +167,7 @@ def make_domain_config(domain, templates, ssl_certificates, env): proxy_redirect_off = False frame_options_header_sameorigin = False web_sockets = False - m = re.search("#(.*)$", url) + m = re.search(r"#(.*)$", url) if m: for flag in m.group(1).split(","): if flag == "pass-http-host": @@ -178,7 +178,7 @@ def make_domain_config(domain, templates, ssl_certificates, env): frame_options_header_sameorigin = True elif flag == "web-sockets": web_sockets = True - url = re.sub("#(.*)$", "", url) + url = re.sub(r"#(.*)$", "", url) nginx_conf_extra += "\tlocation {} {{".format(path) nginx_conf_extra += "\n\t\tproxy_pass {};".format(url) diff --git a/setup/migrate.py b/setup/migrate.py index dbfc54b5..e70343f3 100755 --- a/setup/migrate.py +++ b/setup/migrate.py @@ -23,7 +23,7 @@ def migration_1(env): # Migrate the 'domains' directory. for sslfn in glob.glob(os.path.join( env["STORAGE_ROOT"], 'ssl/domains/*' )): fn = os.path.basename(sslfn) - m = re.match("(.*)_(certifiate.pem|cert_sign_req.csr|private_key.pem)$", fn) + m = re.match(r"(.*)_(certifiate.pem|cert_sign_req.csr|private_key.pem)$", fn) if m: # get the new name for the file domain_name, file_type = m.groups() diff --git a/tests/tls.py b/tests/tls.py index 8795f36a..18b6b7c4 100644 --- a/tests/tls.py +++ b/tests/tls.py @@ -96,7 +96,7 @@ def sslyze(opts, port, ok_ciphers): # Failed. Just output the error. out = re.sub("[\\w\\W]*CHECKING HOST\\(S\\) AVAILABILITY\n\\s*-+\n", "", out) # chop off header that shows the host we queried out = re.sub("[\\w\\W]*SCAN RESULTS FOR.*\n\\s*-+\n", "", out) # chop off header that shows the host we queried - out = re.sub("SCAN COMPLETED IN .*", "", out) + out = re.sub(r"SCAN COMPLETED IN .*", "", out) out = out.rstrip(" \n-") + "\n" # Print. diff --git a/tools/readable_bash.py b/tools/readable_bash.py index ac4b6448..78a4bc08 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -408,7 +408,7 @@ class BashScript(Grammar): string = re.sub(".* #NODOC\n", "", string) string = re.sub("\n\s*if .*then.*|\n\s*fi|\n\s*else|\n\s*elif .*", "", string) string = quasitokenize(string) - string = re.sub("hide_output ", "", string) + string = re.sub(r"hide_output ", "", string) parser = BashScript.parser() result = parser.parse_string(string) From 717e806427e1eb55f243052f20d8e8195a44bff3 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:27:42 -0700 Subject: [PATCH 16/37] Fixed RUF031 (incorrectly-parenthesized-tuple-in-subscript): Avoid parentheses for tuples in subscripts --- management/mail_log.py | 2 +- management/status_checks.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/management/mail_log.py b/management/mail_log.py index 5abacc4f..c9a66939 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -495,7 +495,7 @@ def add_login(user, date, protocol_name, host, collector): data["latest"] = date data["totals_by_protocol"][protocol_name] += 1 - data["totals_by_protocol_and_host"][(protocol_name, host)] += 1 + data["totals_by_protocol_and_host"][protocol_name, host] += 1 if host not in {"127.0.0.1", "::1"} or True: data["activity-by-hour"][protocol_name][date.hour] += 1 diff --git a/management/status_checks.py b/management/status_checks.py index a282be88..284b8dd9 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -620,7 +620,7 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): with open(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/' + dnssec_keys['KSK'] + '.key'), encoding="utf-8") as f: dnsssec_pubkey = f.read().split("\t")[3].split(" ")[3] - expected_ds_records[ (ds_keytag, ds_alg, ds_digalg, ds_digest) ] = { + expected_ds_records[ ds_keytag, ds_alg, ds_digalg, ds_digest ] = { "record": rr_ds, "keytag": ds_keytag, "alg": ds_alg, From 66f140a8cf7d35d48fe808ff525acd4fa2a4dd36 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:27:56 -0700 Subject: [PATCH 17/37] Fixed PGH004 (blanket-noqa): Use a colon when specifying `noqa` rule codes --- management/mail_log.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/management/mail_log.py b/management/mail_log.py index c9a66939..e5509175 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -678,7 +678,7 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None data_accum[col] += d[row] try: - if None not in [latest, earliest]: # noqa PLR6201 + if None not in [latest, earliest]: # noqa: PLR6201 vert_pos = len(line) e = earliest[row] l = latest[row] @@ -731,7 +731,7 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None else: header += l.rjust(max(5, len(l) + 1, col_widths[col])) - if None not in [latest, earliest]: # noqa PLR6201 + if None not in [latest, earliest]: # noqa: PLR6201 header += " │ timespan " lines.insert(0, header.rstrip()) @@ -756,7 +756,7 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None footer += temp.format(data_accum[row]) try: - if None not in [latest, earliest]: # noqa PLR6201 + if None not in [latest, earliest]: # noqa: PLR6201 max_l = max(latest) min_e = min(earliest) timespan = relativedelta(max_l, min_e) From 498e92dc9546c5e01450eb722a53665bb7311c61 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:28:09 -0700 Subject: [PATCH 18/37] Fixed PLW1514 (unspecified-encoding): `open` in text mode without explicit `encoding` argument --- management/mailconfig.py | 2 +- tools/readable_bash.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/management/mailconfig.py b/management/mailconfig.py index 7caee5f7..054b3b0f 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -148,7 +148,7 @@ def get_mail_users_ex(env, with_archived=False): percent = '' try: dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/{}/{}/maildirsize'.format(domain, user)) - with open(dirsize_file, 'r') as f: + with open(dirsize_file, 'r', encoding="utf-8") as f: box_quota = int(f.readline().split('S')[0]) for line in f.readlines(): (size, count) = line.split(' ') diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 78a4bc08..4c79665c 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -124,7 +124,7 @@ def generate_documentation(): """) parser = Source.parser() - with open("setup/start.sh", "r") as start_file: + with open("setup/start.sh", "r", encoding="utf-8") as start_file: for line in start_file: try: fn = parser.parse_string(line).filename() @@ -401,7 +401,7 @@ class BashScript(Grammar): @staticmethod def parse(fn): if fn in ("setup/functions.sh", "/etc/mailinabox.conf"): return "" - with open(fn, "r") as f: + with open(fn, "r", encoding="utf-8") as f: string = f.read() # tokenize From ee11f3849b379f6e93f550eb4ac87a52c28612b1 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:28:17 -0700 Subject: [PATCH 19/37] Fixed UP015 (redundant-open-modes): Unnecessary mode argument --- management/mailconfig.py | 2 +- tools/readable_bash.py | 4 ++-- 2 files changed, 3 insertions(+), 3 deletions(-) diff --git a/management/mailconfig.py b/management/mailconfig.py index 054b3b0f..4024aaaf 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -148,7 +148,7 @@ def get_mail_users_ex(env, with_archived=False): percent = '' try: dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/{}/{}/maildirsize'.format(domain, user)) - with open(dirsize_file, 'r', encoding="utf-8") as f: + with open(dirsize_file, encoding="utf-8") as f: box_quota = int(f.readline().split('S')[0]) for line in f.readlines(): (size, count) = line.split(' ') diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 4c79665c..75a535e2 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -124,7 +124,7 @@ def generate_documentation(): """) parser = Source.parser() - with open("setup/start.sh", "r", encoding="utf-8") as start_file: + with open("setup/start.sh", encoding="utf-8") as start_file: for line in start_file: try: fn = parser.parse_string(line).filename() @@ -401,7 +401,7 @@ class BashScript(Grammar): @staticmethod def parse(fn): if fn in ("setup/functions.sh", "/etc/mailinabox.conf"): return "" - with open(fn, "r", encoding="utf-8") as f: + with open(fn, encoding="utf-8") as f: string = f.read() # tokenize From 213e449dfeca76f228ba1d0c7526bb876a8c11db Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:28:25 -0700 Subject: [PATCH 20/37] Fixed FURB129 (readlines-in-for): Instead of calling `readlines()`, iterate over file object directly --- management/mailconfig.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management/mailconfig.py b/management/mailconfig.py index 4024aaaf..5f0a23fc 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -150,7 +150,7 @@ def get_mail_users_ex(env, with_archived=False): dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/{}/{}/maildirsize'.format(domain, user)) with open(dirsize_file, encoding="utf-8") as f: box_quota = int(f.readline().split('S')[0]) - for line in f.readlines(): + for line in f: (size, count) = line.split(' ') box_size += int(size) From fd2696a42c91966855d01f5dae8efe5d4555f738 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:28:32 -0700 Subject: [PATCH 21/37] Fixed RUF059 (unused-unpacked-variable) --- management/mailconfig.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/management/mailconfig.py b/management/mailconfig.py index 5f0a23fc..853cd7f1 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -151,7 +151,7 @@ def get_mail_users_ex(env, with_archived=False): with open(dirsize_file, encoding="utf-8") as f: box_quota = int(f.readline().split('S')[0]) for line in f: - (size, count) = line.split(' ') + (size, _count) = line.split(' ') box_size += int(size) try: @@ -389,7 +389,7 @@ def hash_password(pw): def get_mail_quota(email, env): - conn, c = open_database(env, with_connection=True) + _conn, c = open_database(env, with_connection=True) c.execute("SELECT quota FROM users WHERE email=?", (email,)) rows = c.fetchall() if len(rows) != 1: From c3826e45aa9a624a5e0a576d77fb3845452917bb Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:28:58 -0700 Subject: [PATCH 22/37] Fixed SIM101 (duplicate-isinstance-call): Multiple `isinstance` calls for `pem`, merge into a single call --- management/ssl_certificates.py | 4 +--- 1 file changed, 1 insertion(+), 3 deletions(-) diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index effb6ace..f7d481a8 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -63,9 +63,7 @@ def get_ssl_certificates(env): if isinstance(pem, Certificate): certificates.append({ "filename": fn, "cert": pem }) # It is a private key - elif (isinstance(pem, rsa.RSAPrivateKey) - or isinstance(pem, dsa.DSAPrivateKey) - or isinstance(pem, ec.EllipticCurvePrivateKey)): + elif (isinstance(pem, (rsa.RSAPrivateKey, dsa.DSAPrivateKey, ec.EllipticCurvePrivateKey))): private_keys[pem.public_key().public_numbers()] = { "filename": fn, "key": pem } From 04ed7529489985daeff607f32330a2f8c8edda3b Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:29:27 -0700 Subject: [PATCH 23/37] Fixed FURB142 (for-loop-set-mutations): Use of `set.add()` in a for loop --- management/ssl_certificates.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index f7d481a8..ce72cc6d 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -671,8 +671,7 @@ def get_certificate_domains(cert): try: sans = cert.extensions.get_extension_for_oid(OID_SUBJECT_ALTERNATIVE_NAME).value.get_values_for_type(DNSName) - for san in sans: - names.add(idna_decode_dns_name(san)) + names.update(idna_decode_dns_name(san) for san in sans) except ExtensionNotFound: pass From 562f76e61fd6ebc809ba14ff5b96036b02c1152d Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:29:46 -0700 Subject: [PATCH 24/37] Fixed UP032 (f-string): Use f-string instead of `format` call --- management/cli.py | 2 +- management/daemon.py | 2 +- management/dns_update.py | 32 ++++++++++++++++---------------- management/mail_log.py | 10 +++++----- management/mailconfig.py | 32 ++++++++++++++++---------------- management/ssl_certificates.py | 10 +++++----- management/status_checks.py | 28 ++++++++++++++-------------- management/web_update.py | 10 +++++----- setup/migrate.py | 2 +- tests/fail2ban.py | 2 +- tests/test_dns.py | 6 +++--- tests/test_mail.py | 4 ++-- tools/readable_bash.py | 6 +++--- 13 files changed, 73 insertions(+), 73 deletions(-) diff --git a/management/cli.py b/management/cli.py index f8089607..f11877d7 100755 --- a/management/cli.py +++ b/management/cli.py @@ -120,7 +120,7 @@ elif sys.argv[1] == "user" and sys.argv[2] == "admins": elif sys.argv[1] == "user" and sys.argv[2] == "quota" and len(sys.argv) == 4: # Get a user's quota - print(mgmt("/mail/users/quota?text=1&email={}".format(sys.argv[3]))) + print(mgmt(f"/mail/users/quota?text=1&email={sys.argv[3]}")) elif sys.argv[1] == "user" and sys.argv[2] == "quota" and len(sys.argv) == 5: # Set a user's quota diff --git a/management/daemon.py b/management/daemon.py index 7b66f196..a398d0aa 100755 --- a/management/daemon.py +++ b/management/daemon.py @@ -736,7 +736,7 @@ def munin_cgi(filename): query_str = request.query_string.decode("utf-8", 'ignore') - env = {'PATH_INFO': '/{}/'.format(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 diff --git a/management/dns_update.py b/management/dns_update.py index dbb25a4e..840b98ad 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -252,16 +252,16 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) # 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 {} resolves to for web hosting and other services besides mail. The A record must be present but its value does not affect mail delivery.".format(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 {} resolves to so that the box can provide a redirect to the parent domain.".format(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 {} resolves to, e.g. for web hosting. (It is not necessary for receiving mail on this domain.)".format(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 @@ -279,13 +279,13 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) 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 {}.".format(env["PRIMARY_HOSTNAME"]), "Required. Specifies the hostname (and priority) of the machine that handles @{} mail.".format(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 @{} mail.".format(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. @@ -294,12 +294,12 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) 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 @{} mail.".format(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 @{} or which does not have a valid DKIM signature is suspect and should be quarantined by the recipient's mail system.".format(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 @@ -362,9 +362,9 @@ def build_zone(domain, domain_properties, additional_records, env, is_zone=True) # 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 @{}. 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).".format(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 @{}.".format(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"): @@ -590,7 +590,7 @@ def get_dns_zonefile(zone, env): if zone == domain: break else: - raise ValueError("{} is not a domain name that corresponds to a zone.".format(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: @@ -615,8 +615,8 @@ zone: # 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: {} NOKEY".format(ipaddr) - nsdconf += "\n\tprovide-xfr: {} NOKEY\n".format(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. @@ -896,7 +896,7 @@ def set_custom_dns_record(qname, rtype, value, action, env): else: # No match. if qname != "_secondary_nameserver": - raise ValueError("{} is not a domain name or a subdomain of a domain name managed by this box.".format(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() @@ -926,7 +926,7 @@ def set_custom_dns_record(qname, rtype, value, action, env): # anything goes pass else: - raise ValueError("Unknown record type '{}'.".format(rtype)) + raise ValueError(f"Unknown record type '{rtype}'.") # load existing config config = list(get_custom_dns_config(env)) @@ -1037,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 {}.".format(item)) + raise ValueError(f"Could not resolve the IP address of {item}.") else: # Validate IP address. try: @@ -1046,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("'{}' is not an IPv4 or IPv6 address or subnet.".format(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) diff --git a/management/mail_log.py b/management/mail_log.py index e5509175..60e5e649 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -302,7 +302,7 @@ def scan_mail_log(env): for date, sender, message in user_data["blocked"]: if len(sender) > 64: sender = sender[:32] + "…" + sender[-32:] - user_rejects.extend((f'{date} - {sender} ', ' {}'.format(message))) + user_rejects.extend((f'{date} - {sender} ', f' {message}')) rejects.append(user_rejects) print_user_table( @@ -608,7 +608,7 @@ def valid_date(string): try: date = dateutil.parser.parse(string) except ValueError: - raise argparse.ArgumentTypeError("Unrecognized date and/or time '{}'".format(string)) + raise argparse.ArgumentTypeError(f"Unrecognized date and/or time '{string}'") return date @@ -669,7 +669,7 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None col_str = f"{d[row]!s:<20}" col_left[col] = True else: - temp = "{{:>{}}}".format(max(5, len(l) + 1, len(str(d[row])) + 1)) + temp = f"{{:>{max(5, len(l) + 1, len(str(d[row])) + 1)}}}" col_str = temp.format(str(d[row])) col_widths[col] = max(col_widths[col], len(col_str)) line += col_str @@ -706,10 +706,10 @@ def print_user_table(users, data=None, sub_data=None, activity=None, latest=None if sub_data is not None: for l, d in sub_data: if d[row]: - lines.extend(('┬', '│ {}'.format(l), '├─%s─' % (len(l) * '─'), '│')) + lines.extend(('┬', f'│ {l}', '├─%s─' % (len(l) * '─'), '│')) max_len = 0 for v in list(d[row]): - lines.append("│ {}".format(v)) + lines.append(f"│ {v}") max_len = max(max_len, len(v)) lines.append("└" + (max_len + 1) * "─") diff --git a/management/mailconfig.py b/management/mailconfig.py index 853cd7f1..22c18fff 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -108,8 +108,8 @@ def sizeof_fmt(num): for unit in ['','K','M','G','T']: if abs(num) < 1024.0: if abs(num) > 99: - return "{:3.0f}{}".format(num, unit) - return "{:2.1f}{}".format(num, unit) + return f"{num:3.0f}{unit}" + return f"{num:2.1f}{unit}" num /= 1024.0 @@ -147,7 +147,7 @@ def get_mail_users_ex(env, with_archived=False): box_quota = 0 percent = '' try: - dirsize_file = os.path.join(env['STORAGE_ROOT'], 'mail/mailboxes/{}/{}/maildirsize'.format(domain, user)) + dirsize_file = os.path.join(env['STORAGE_ROOT'], f'mail/mailboxes/{domain}/{user}/maildirsize') with open(dirsize_file, encoding="utf-8") as f: box_quota = int(f.readline().split('S')[0]) for line in f: @@ -173,7 +173,7 @@ def get_mail_users_ex(env, with_archived=False): "quota": quota, "box_quota": box_quota, "box_size": sizeof_fmt(box_size) if box_size != '?' else box_size, - "percent": '{:3.0f}%'.format(percent) if type(percent) != str else percent, + "percent": f'{percent:3.0f}%' if type(percent) != str else percent, "status": "active", } users.append(user) @@ -377,7 +377,7 @@ def set_mail_password(email, pw, env): conn, c = open_database(env, with_connection=True) c.execute("UPDATE users SET password=? WHERE email=?", (pw, email)) if c.rowcount != 1: - return ("That's not a user ({}).".format(email), 400) + return (f"That's not a user ({email}).", 400) conn.commit() return "OK" @@ -393,7 +393,7 @@ def get_mail_quota(email, env): c.execute("SELECT quota FROM users WHERE email=?", (email,)) rows = c.fetchall() if len(rows) != 1: - return ("That's not a user ({}).".format(email), 400) + return (f"That's not a user ({email}).", 400) return rows[0][0] @@ -406,7 +406,7 @@ def set_mail_quota(email, quota, env): conn, c = open_database(env, with_connection=True) c.execute("UPDATE users SET quota=? WHERE email=?", (quota, email)) if c.rowcount != 1: - return ("That's not a user ({}).".format(email), 400) + return (f"That's not a user ({email}).", 400) conn.commit() dovecot_quota_recalc(email) @@ -448,7 +448,7 @@ def get_mail_password(email, env): c.execute('SELECT password FROM users WHERE email=?', (email,)) rows = c.fetchall() if len(rows) != 1: - raise ValueError("That's not a user ({}).".format(email)) + raise ValueError(f"That's not a user ({email}).") return rows[0][0] def remove_mail_user(email, env): @@ -456,7 +456,7 @@ def remove_mail_user(email, env): conn, c = open_database(env, with_connection=True) c.execute("DELETE FROM users WHERE email=?", (email,)) if c.rowcount != 1: - return ("That's not a user ({}).".format(email), 400) + return (f"That's not a user ({email}).", 400) conn.commit() # Update things in case any domains are removed. @@ -472,12 +472,12 @@ def get_mail_user_privileges(email, env, empty_on_error=False): rows = c.fetchall() if len(rows) != 1: if empty_on_error: return [] - return ("That's not a user ({}).".format(email), 400) + return (f"That's not a user ({email}).", 400) return parse_privs(rows[0][0]) def validate_privilege(priv): if "\n" in priv or priv.strip() == "": - return ("That's not a valid privilege ({}).".format(priv), 400) + return (f"That's not a valid privilege ({priv}).", 400) return None def add_remove_mail_user_privilege(email, priv, action, env): @@ -520,7 +520,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist if address == "": return ("No email address provided.", 400) if not validate_email(address, mode='alias'): - return ("Invalid email address ({}).".format(address), 400) + return (f"Invalid email address ({address}).", 400) # validate forwards_to validated_forwards_to = [] @@ -549,7 +549,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist # Strip any +tag from email alias and check privileges privileged_email = re.sub(r"(?=\+)[^@]*(?=@)",'',email) if not validate_email(email): - return ("Invalid receiver email address ({}).".format(email), 400) + return (f"Invalid receiver email address ({email}).", 400) if is_dcv_source and not is_dcv_address(email) and "admin" not in get_mail_user_privileges(privileged_email, env, empty_on_error=True): # Make domain control validation hijacking a little harder to mess up by # requiring aliases for email addresses typically used in DCV to forward @@ -569,7 +569,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist login = login.strip() if login == "": continue if login not in valid_logins: - return ("Invalid permitted sender: {} is not a user on this system.".format(login), 400) + return (f"Invalid permitted sender: {login} is not a user on this system.", 400) validated_permitted_senders.append(login) # Make sure the alias has either a forwards_to or a permitted_sender. @@ -588,7 +588,7 @@ def add_mail_alias(address, forwards_to, permitted_senders, env, update_if_exist return_status = "alias added" except sqlite3.IntegrityError: if not update_if_exists: - return ("Alias already exists ({}).".format(address), 400) + return (f"Alias already exists ({address}).", 400) c.execute("UPDATE aliases SET destination = ?, permitted_senders = ? WHERE source = ?", (forwards_to, permitted_senders, address)) return_status = "alias updated" @@ -607,7 +607,7 @@ def remove_mail_alias(address, env, do_kick=True): conn, c = open_database(env, with_connection=True) c.execute("DELETE FROM aliases WHERE source=?", (address,)) if c.rowcount != 1: - return ("That's not an alias ({}).".format(address), 400) + return (f"That's not an alias ({address}).", 400) conn.commit() if do_kick: diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index ce72cc6d..5b9470ec 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -515,7 +515,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring cert = load_pem(ssl_cert_chain[0]) if not isinstance(cert, Certificate): raise ValueError("This is not a certificate file.") except ValueError as e: - return ("There is a problem with the certificate file: {}".format(str(e)), None) + return (f"There is a problem with the certificate file: {str(e)}", None) # First check that the domain name is one of the names allowed by # the certificate. @@ -540,10 +540,10 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring if (not isinstance(priv_key, rsa.RSAPrivateKey) and not isinstance(priv_key, dsa.DSAPrivateKey) and not isinstance(priv_key, ec.EllipticCurvePrivateKey)): - return ("The private key file {} is not a private key file.".format(ssl_private_key), None) + return (f"The private key file {ssl_private_key} is not a private key file.", None) if priv_key.public_key().public_numbers() != cert.public_key().public_numbers(): - return ("The certificate does not correspond to the private key at {}.".format(ssl_private_key), None) + return (f"The certificate does not correspond to the private key at {ssl_private_key}.", None) # We could also use the openssl command line tool to get the modulus # listed in each file. The output of each command below looks like "Modulus=XXXXX". @@ -589,7 +589,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring if retcode != 0: if "unable to get local issuer certificate" in verifyoutput: - return ("The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. ({})".format(verifyoutput), None) + return (f"The certificate is missing an intermediate chain or the intermediate chain is incorrect or incomplete. ({verifyoutput})", None) # There is some unknown problem. Return the `openssl verify` raw output. return ("There is a problem with the certificate.", verifyoutput.strip()) @@ -605,7 +605,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring expiry_info = "The certificate expires in %d days on %s." % (ndays, cert_expiration_date.date().isoformat()) else: # We'll renew it with Lets Encrypt. - expiry_info = "The certificate expires on {}.".format(cert_expiration_date.date().isoformat()) + expiry_info = f"The certificate expires on {cert_expiration_date.date().isoformat()}." if warn_if_expiring_soon and ndays <= warn_if_expiring_soon: # Warn on day 10 to give 4 days for us to automatically renew the diff --git a/management/status_checks.py b/management/status_checks.py index 284b8dd9..da2c2f16 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -248,7 +248,7 @@ def check_free_disk_space(rounded_values, env, output): def check_free_memory(rounded_values, env, output): # Check free memory. percent_free = 100 - psutil.virtual_memory().percent - memory_msg = "System memory is {}% free.".format(str(round(percent_free))) + memory_msg = f"System memory is {str(round(percent_free))}% free." if percent_free >= 20: if rounded_values: memory_msg = "System free memory is at least 20%." output.print_ok(memory_msg) @@ -478,7 +478,7 @@ def check_primary_hostname_dns(domain, env, output, dns_domains, dns_zonefiles): tlsa25 = query_dns(tlsa_qname, "TLSA", nxdomain=None) tlsa25_expected = build_tlsa_record(env) if tlsa25 == tlsa25_expected: - output.print_ok("""The DANE TLSA record for incoming mail is correct ({}).""".format(tlsa_qname),) + output.print_ok(f"""The DANE TLSA record for incoming mail is correct ({tlsa_qname}).""",) elif tlsa25 is None: if has_dnssec: # Omit a warning about it not being set if DNSSEC isn't enabled, @@ -497,9 +497,9 @@ def check_alias_exists(alias_name, alias, env, output): if mail_aliases[alias]: output.print_ok(f"{alias_name} exists as a mail alias. [{alias} ↦ {mail_aliases[alias]}]") else: - output.print_error("""You must set the destination of the mail alias for {} to direct email to you or another administrator.""".format(alias)) + output.print_error(f"""You must set the destination of the mail alias for {alias} to direct email to you or another administrator.""") else: - output.print_error("""You must add a mail alias for {} which directs email to you or another administrator.""".format(alias)) + output.print_error(f"""You must add a mail alias for {alias} which directs email to you or another administrator.""") def check_dns_zone(domain, env, output, dns_zonefiles): # If a DS record is set at the registrar, check DNSSEC first because it will affect the NS query. @@ -527,7 +527,7 @@ def check_dns_zone(domain, env, output, dns_zonefiles): probably_external_dns = False if existing_ns.lower() == correct_ns.lower(): - output.print_ok("Nameservers are set correctly at registrar. [{}]".format(correct_ns)) + output.print_ok(f"Nameservers are set correctly at registrar. [{correct_ns}]") elif ip == correct_ip: # The domain resolves correctly, so maybe the user is using External DNS. output.print_warning(f"""The nameservers set on this domain at your domain name registrar should be {correct_ns}. They are currently {existing_ns}. @@ -546,7 +546,7 @@ def check_dns_zone(domain, env, output, dns_zonefiles): # We must first resolve the nameserver to an IP address so we can query it. ns_ips = query_dns(ns, "A") if not ns_ips or ns_ips in {'[Not Set]', '[timeout]'}: - output.print_error("Secondary nameserver {} is not valid (it doesn't resolve to an IP address).".format(ns)) + output.print_error(f"Secondary nameserver {ns} is not valid (it doesn't resolve to an IP address).") continue # Choose the first IP if nameserver returns multiple ns_ip = ns_ips.split('; ')[0] @@ -587,7 +587,7 @@ def check_dns_zone_suggestions(domain, env, output, dns_zonefiles, domains_with_ if domain in domains_with_a_records: output.print_warning("""Web has been disabled for this domain because you have set a custom DNS record.""") if "www." + domain in domains_with_a_records: - output.print_warning("""A redirect from 'www.{}' has been disabled for this domain because you have set a custom DNS record on the www subdomain.""".format(domain)) + output.print_warning(f"""A redirect from 'www.{domain}' has been disabled for this domain because you have set a custom DNS record on the www subdomain.""") # Since DNSSEC is optional, if a DS record is NOT set at the registrar suggest it. # (If it was set, we did the check earlier.) @@ -616,7 +616,7 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): # Some registrars may want the public key so they can compute the digest. The DS # record that we suggest using is for the KSK (and that's how the DS records were generated). # We'll also give the nice name for the key algorithm. - dnssec_keys = load_env_vars_from_file(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/{}.conf'.format(alg_name_map[ds_alg]))) + dnssec_keys = load_env_vars_from_file(os.path.join(env['STORAGE_ROOT'], f'dns/dnssec/{alg_name_map[ds_alg]}.conf')) with open(os.path.join(env['STORAGE_ROOT'], 'dns/dnssec/' + dnssec_keys['KSK'] + '.key'), encoding="utf-8") as f: dnsssec_pubkey = f.read().split("\t")[3].split(" ")[3] @@ -662,7 +662,7 @@ def check_dnssec(domain, env, output, dns_zonefiles, is_checking_primary=False): for this domain is valid.""") else: if is_checking_primary: - output.print_error("""The DNSSEC 'DS' record for {} is incorrect. See further details below.""".format(domain)) + output.print_error(f"""The DNSSEC 'DS' record for {domain} is incorrect. See further details below.""") return output.print_error("""This domain's DNSSEC DS record is incorrect. The chain of trust is broken between the public DNS system and this machine's DNS server. It may take several hours for public DNS to update after a change. If you did not recently @@ -776,9 +776,9 @@ def check_mail_domain(domain, env, output): elif dbl == "127.255.255.252": output.print_warning("Incorrect spamhaus query: {}. Could not determine whether the domain {} is blacklisted.".format(domain+'.dbl.spamhaus.org', domain)) elif dbl == "127.255.255.254": - output.print_warning("Mail-in-a-Box is configured to use a public DNS server. This is not supported by spamhaus. Could not determine whether the domain {} is blacklisted.".format(domain)) + output.print_warning(f"Mail-in-a-Box is configured to use a public DNS server. This is not supported by spamhaus. Could not determine whether the domain {domain} is blacklisted.") elif dbl == "127.255.255.255": - output.print_warning("Too many queries have been performed on the spamhaus server. Could not determine whether the domain {} is blacklisted.".format(domain)) + output.print_warning(f"Too many queries have been performed on the spamhaus server. Could not determine whether the domain {domain} is blacklisted.") else: output.print_error(f"""This domain is listed in the Spamhaus Domain Block List (code {dbl}), which may prevent recipients from receiving your mail. @@ -960,14 +960,14 @@ def check_miab_version(env, output): this_ver = "Unknown" if config.get("privacy", True): - output.print_warning("You are running version Mail-in-a-Box {}. Mail-in-a-Box version check disabled by privacy setting.".format(this_ver)) + output.print_warning(f"You are running version Mail-in-a-Box {this_ver}. Mail-in-a-Box version check disabled by privacy setting.") else: latest_ver = get_latest_miab_version() if this_ver == latest_ver: - output.print_ok("Mail-in-a-Box is up to date. You are running version {}.".format(this_ver)) + output.print_ok(f"Mail-in-a-Box is up to date. You are running version {this_ver}.") elif latest_ver is None: - output.print_error("Latest Mail-in-a-Box version could not be determined. You are running version {}.".format(this_ver)) + output.print_error(f"Latest Mail-in-a-Box version could not be determined. You are running version {this_ver}.") else: output.print_error(f"A new version of Mail-in-a-Box is available. You are running version {this_ver}. The latest version is {latest_ver}. For upgrade instructions, see https://mailinabox.email. ") diff --git a/management/web_update.py b/management/web_update.py index 2d7ec0b2..7ed99dbf 100644 --- a/management/web_update.py +++ b/management/web_update.py @@ -180,8 +180,8 @@ def make_domain_config(domain, templates, ssl_certificates, env): web_sockets = True url = re.sub(r"#(.*)$", "", url) - nginx_conf_extra += "\tlocation {} {{".format(path) - nginx_conf_extra += "\n\t\tproxy_pass {};".format(url) + nginx_conf_extra += f"\tlocation {path} {{" + nginx_conf_extra += f"\n\t\tproxy_pass {url};" if proxy_redirect_off: nginx_conf_extra += "\n\t\tproxy_redirect off;" if pass_http_host_header: @@ -198,8 +198,8 @@ def make_domain_config(domain, templates, ssl_certificates, env): nginx_conf_extra += "\n\t\tproxy_set_header X-Real-IP $remote_addr;" nginx_conf_extra += "\n\t}\n" for path, alias in yaml.get("aliases", {}).items(): - nginx_conf_extra += "\tlocation {} {{".format(path) - nginx_conf_extra += "\n\t\talias {};".format(alias) + nginx_conf_extra += f"\tlocation {path} {{" + nginx_conf_extra += f"\n\t\talias {alias};" nginx_conf_extra += "\n\t}\n" for path, url in yaml.get("redirects", {}).items(): nginx_conf_extra += f"\trewrite {path} {url} permanent;\n" @@ -216,7 +216,7 @@ def make_domain_config(domain, templates, ssl_certificates, env): # Add in any user customizations in the includes/ folder. nginx_conf_custom_include = os.path.join(env["STORAGE_ROOT"], "www", safe_domain_name(domain) + ".conf") if os.path.exists(nginx_conf_custom_include): - nginx_conf_extra += "\tinclude {};\n".format(nginx_conf_custom_include) + nginx_conf_extra += f"\tinclude {nginx_conf_custom_include};\n" # PUT IT ALL TOGETHER # Combine the pieces. Iteratively place each template into the "# ADDITIONAL DIRECTIVES HERE" placeholder diff --git a/setup/migrate.py b/setup/migrate.py index e70343f3..422179ca 100755 --- a/setup/migrate.py +++ b/setup/migrate.py @@ -164,7 +164,7 @@ def migration_12(env): try: table = table[0] c = conn.cursor() - dropcmd = "DROP TABLE {}".format(table) + dropcmd = f"DROP TABLE {table}" c.execute(dropcmd) except: print("Failed to drop table", table) diff --git a/tests/fail2ban.py b/tests/fail2ban.py index abddcc41..30e3c267 100644 --- a/tests/fail2ban.py +++ b/tests/fail2ban.py @@ -142,7 +142,7 @@ def http_test(url, expected_status, postdata=None, qsargs=None, auth=None): # return response status code if r.status_code != expected_status: r.raise_for_status() # anything but 200 - raise OSError("Got unexpected status code {}.".format(r.status_code)) + raise OSError(f"Got unexpected status code {r.status_code}.") # define how to run a test diff --git a/tests/test_dns.py b/tests/test_dns.py index 6f84d5d0..f834b744 100755 --- a/tests/test_dns.py +++ b/tests/test_dns.py @@ -51,7 +51,7 @@ def test2(tests, server, description): response = dns.resolver.resolve(qname, rtype) except dns.resolver.NoNameservers: # host did not have an answer for this query - print("Could not connect to {} for DNS query.".format(server)) + print(f"Could not connect to {server} for DNS query.") sys.exit(1) except (dns.resolver.NXDOMAIN, dns.resolver.NoAnswer): # host did not have an answer for this query; not sure what the @@ -79,7 +79,7 @@ def test2(tests, server, description): # Test the response from the machine itself. if not test(ipaddr, "Mail-in-a-Box"): print () - print ("Please run the Mail-in-a-Box setup script on {} again.".format(hostname)) + print (f"Please run the Mail-in-a-Box setup script on {hostname} again.") sys.exit(1) else: print ("The Mail-in-a-Box provided correct DNS answers.") @@ -89,7 +89,7 @@ else: # to see if the machine is hooked up to recursive DNS properly. if not test("8.8.8.8", "Google Public DNS"): print () - print ("Check that the nameserver settings for {} are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box.".format(hostname)) + print (f"Check that the nameserver settings for {hostname} are correct at your domain registrar. It may take a few hours for Google Public DNS to update after changes on your Mail-in-a-Box.") sys.exit(1) else: print ("Your domain registrar or DNS host appears to be configured correctly as well. Public DNS provides the same answers.") diff --git a/tests/test_mail.py b/tests/test_mail.py index 9cc7bb03..3a5de766 100755 --- a/tests/test_mail.py +++ b/tests/test_mail.py @@ -46,7 +46,7 @@ reverse_ip = dns.reversename.from_address(ipaddr) # e.g. "1.0.0.127.in-addr.arpa try: reverse_dns = dns.resolver.resolve(reverse_ip, 'PTR')[0].target.to_text(omit_final_dot=True) # => hostname except dns.resolver.NXDOMAIN: - print("Reverse DNS lookup failed for {}. SMTP EHLO name check skipped.".format(ipaddr)) + print(f"Reverse DNS lookup failed for {ipaddr}. SMTP EHLO name check skipped.") reverse_dns = None if reverse_dns is not None: server.ehlo_or_helo_if_needed() # must send EHLO before getting the server's EHLO name @@ -54,7 +54,7 @@ if reverse_dns is not None: if helo_name != reverse_dns: print("The server's EHLO name does not match its reverse hostname. Check DNS settings.") else: - print("SMTP EHLO name ({}) is OK.".format(helo_name)) + print(f"SMTP EHLO name ({helo_name}) is OK.") # Login and send a test email. server.login(emailaddress, pw) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 75a535e2..54703339 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -229,7 +229,7 @@ class EditConf(Grammar): for opt in re.split("\s+", self[4].string): k, v = opt.split("=", 1) v = re.sub(r"\n+", "", fixup_tokens(v)) # not sure why newlines are getting doubled - options.append("{}{}{}".format(k, eq, v)) + options.append(f"{k}{eq}{v}") return "
" + self[1].string + " (change settings)
" + "\n".join(cgi.escape(s) for s in options) + "
\n" class CaptureOutput(Grammar): @@ -247,7 +247,7 @@ class SedReplace(Grammar): class EchoPipe(Grammar): grammar = OPTIONAL(SPACE), L("echo "), REST_OF_LINE, L(' | '), REST_OF_LINE, EOL def value(self): - text = " ".join("\"{}\"".format(s) for s in self[2].string.split(" ")) + text = " ".join(f"\"{s}\"" for s in self[2].string.split(" ")) return "
echo " + recode_bash(text) + " \
| " + recode_bash(self[4].string) + "
\n" def shell_line(bash): @@ -427,7 +427,7 @@ class BashScript(Grammar): mode = 0 clz = "contd" if mode == 0: - v += "
\n".format(clz) + v += f"
\n" v += "
\n" v += item mode = 1 From b546ccd1625760d1ca7f5c80ea4e8daaa7838107 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:30:08 -0700 Subject: [PATCH 25/37] Fixed FURB110 (if-exp-instead-of-or-operator): Replace ternary `if` expression with `or` operator --- management/status_checks.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management/status_checks.py b/management/status_checks.py index da2c2f16..b584f9e1 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -1114,7 +1114,7 @@ class ConsoleOutput(FileOutput): class BufferedOutput: # Record all of the instance method calls so we can play them back later. def __init__(self, with_lines=None): - self.buf = with_lines if with_lines else [] + self.buf = with_lines or [] def __getattr__(self, attr): if attr not in {"add_heading", "print_ok", "print_error", "print_warning", "print_block", "print_line"}: raise AttributeError From 5c30299461c3e8c26bf573f20271b4b0535fd6ed Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:30:18 -0700 Subject: [PATCH 26/37] Fixed FURB122 (for-loop-writes): Use of `f.write` in a for loop --- management/utils.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/management/utils.py b/management/utils.py index 1a5b2195..fbc4483a 100644 --- a/management/utils.py +++ b/management/utils.py @@ -21,8 +21,7 @@ def load_env_vars_from_file(fn): def save_environment(env): with open("/etc/mailinabox.conf", "w", encoding="utf-8") as f: - for k, v in env.items(): - f.write(f"{k}={v}\n") + f.writelines(f"{k}={v}\n" for k, v in env.items()) # THE SETTINGS FILE AT STORAGE_ROOT/settings.yaml. From 89e4adcfb59497799673cbaf004636220485ccd0 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:31:08 -0700 Subject: [PATCH 27/37] Fixed PLR6201 (literal-membership): Use a set literal when testing for membership --- tools/readable_bash.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 54703339..85bbaeb0 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -130,7 +130,7 @@ def generate_documentation(): fn = parser.parse_string(line).filename() except: continue - if fn in ("setup/start.sh", "setup/preflight.sh", "setup/questions.sh", "setup/firstuser.sh", "setup/management.sh"): + if fn in {"setup/start.sh", "setup/preflight.sh", "setup/questions.sh", "setup/firstuser.sh", "setup/management.sh"}: continue import sys @@ -323,7 +323,7 @@ def quasitokenize(bashscript): elif c == "\\": # Escaping next character. escape_next = True - elif quote_mode is None and c in ('"', "'"): + elif quote_mode is None and c in {'"', "'"}: # Starting a quoted word. quote_mode = c elif c == quote_mode: @@ -400,7 +400,7 @@ class BashScript(Grammar): @staticmethod def parse(fn): - if fn in ("setup/functions.sh", "/etc/mailinabox.conf"): return "" + if fn in {"setup/functions.sh", "/etc/mailinabox.conf"}: return "" with open(fn, encoding="utf-8") as f: string = f.read() From 560677085e0d082ed61f85f0a7c66d80dc3403f7 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:32:13 -0700 Subject: [PATCH 28/37] Fixed F841 (unused-variable): Local variable `conffile` is assigned to but never used --- tools/readable_bash.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 85bbaeb0..eef96595 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -222,7 +222,7 @@ class EditConf(Grammar): EOL ) def value(self): - conffile = self[1] + # conffile = self[1] options = [] eq = "=" if self[3] and "-s" in self[3].string: eq = " " From 8aef7aef64e61c5e1d1c7730da73daaf33c71391 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:32:21 -0700 Subject: [PATCH 29/37] Fixed W605 (invalid-escape-sequence) --- tools/readable_bash.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index eef96595..290fea1e 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -226,7 +226,7 @@ class EditConf(Grammar): options = [] eq = "=" if self[3] and "-s" in self[3].string: eq = " " - for opt in re.split("\s+", self[4].string): + for opt in re.split(r"\s+", self[4].string): k, v = opt.split("=", 1) v = re.sub(r"\n+", "", fixup_tokens(v)) # not sure why newlines are getting doubled options.append(f"{k}{eq}{v}") @@ -248,7 +248,7 @@ class EchoPipe(Grammar): grammar = OPTIONAL(SPACE), L("echo "), REST_OF_LINE, L(' | '), REST_OF_LINE, EOL def value(self): text = " ".join(f"\"{s}\"" for s in self[2].string.split(" ")) - return "
echo " + recode_bash(text) + " \
| " + recode_bash(self[4].string) + "
\n" + return "
echo " + recode_bash(text) + r" \
| " + recode_bash(self[4].string) + "
\n" def shell_line(bash): return "
" + recode_bash(bash.strip()) + "
\n" @@ -363,7 +363,7 @@ def quasitokenize(bashscript): newscript += c # "<< EOF" escaping. - if quote_mode is None and re.search("<<\s*EOF\n$", newscript): + if quote_mode is None and re.search("<<\\s*EOF\n$", newscript): quote_mode = "EOF" elif quote_mode == "EOF" and re.search("\nEOF\n$", newscript): quote_mode = None @@ -406,7 +406,7 @@ class BashScript(Grammar): # tokenize string = re.sub(".* #NODOC\n", "", string) - string = re.sub("\n\s*if .*then.*|\n\s*fi|\n\s*else|\n\s*elif .*", "", string) + string = re.sub("\n\\s*if .*then.*|\n\\s*fi|\n\\s*else|\n\\s*elif .*", "", string) string = quasitokenize(string) string = re.sub(r"hide_output ", "", string) @@ -458,7 +458,7 @@ class BashScript(Grammar): v = fixup_tokens(v) v = v.replace("\n
", "")
-		v = re.sub("
([\w\W]*?)
", lambda m : "
" + strip_indent(m.group(1)) + "
", v) + v = re.sub(r"
([\w\W]*?)
", lambda m : "
" + strip_indent(m.group(1)) + "
", v) v = re.sub(r"(\$?)PRIMARY_HOSTNAME", r"box.yourdomain.com", v) v = re.sub(r"\$STORAGE_ROOT", r"$STORE", v) @@ -468,7 +468,7 @@ class BashScript(Grammar): def wrap_lines(text, cols=60): ret = "" - words = re.split("(\s+)", text) + words = re.split(r"(\s+)", text) linelen = 0 for w in words: if linelen + len(w) > cols-1: From ed1579a5c6c1a87e87adc4221213423952d55265 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:32:29 -0700 Subject: [PATCH 30/37] Fixed Q003 (avoidable-escaped-quote): Change outer quotes to avoid escaping inner quotes --- tools/readable_bash.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 290fea1e..a97fcf8c 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -247,7 +247,7 @@ class SedReplace(Grammar): class EchoPipe(Grammar): grammar = OPTIONAL(SPACE), L("echo "), REST_OF_LINE, L(' | '), REST_OF_LINE, EOL def value(self): - text = " ".join(f"\"{s}\"" for s in self[2].string.split(" ")) + text = " ".join(f'"{s}"' for s in self[2].string.split(" ")) return "
echo " + recode_bash(text) + r" \
| " + recode_bash(self[4].string) + "
\n" def shell_line(bash): @@ -377,7 +377,7 @@ def recode_bash(s): tok = tok.replace(c, "\\" + c) tok = fixup_tokens(tok) if " " in tok or '"' in tok: - tok = tok.replace("\"", "\\\"") + tok = tok.replace('"', '\\"') tok = '"' + tok +'"' else: tok = tok.replace("'", "\\'") From 529c7e6dd5f815f961d2d6aad66bbc7e23559d32 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:32:38 -0700 Subject: [PATCH 31/37] Fixed RUF055 (unnecessary-regular-expression): Plain string pattern passed to `re` function --- tools/readable_bash.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index a97fcf8c..51da28dc 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -408,7 +408,7 @@ class BashScript(Grammar): string = re.sub(".* #NODOC\n", "", string) string = re.sub("\n\\s*if .*then.*|\n\\s*fi|\n\\s*else|\n\\s*elif .*", "", string) string = quasitokenize(string) - string = re.sub(r"hide_output ", "", string) + string = string.replace(r"hide_output ", "") parser = BashScript.parser() result = parser.parse_string(string) From 5c9c1705d075ece85a7a1056b82dba063a1c451e Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:32:46 -0700 Subject: [PATCH 32/37] Fixed RET504 (unnecessary-assign): Unnecessary assignment to `v` before `return` statement --- tools/readable_bash.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/tools/readable_bash.py b/tools/readable_bash.py index 51da28dc..fc21faa9 100644 --- a/tools/readable_bash.py +++ b/tools/readable_bash.py @@ -462,9 +462,8 @@ class BashScript(Grammar): v = re.sub(r"(\$?)PRIMARY_HOSTNAME", r"box.yourdomain.com", v) v = re.sub(r"\$STORAGE_ROOT", r"$STORE", v) - v = v.replace("`pwd`", "/path/to/mailinabox") + return v.replace("`pwd`", "/path/to/mailinabox") - return v def wrap_lines(text, cols=60): ret = "" From 54750b1763688ff66e9ebbd465f0dceba4a9db1b Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:36:22 -0700 Subject: [PATCH 33/37] Fixed RET506 (superfluous-else-raise): Unnecessary `elif` after `raise` statement --- management/backup.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/management/backup.py b/management/backup.py index ecd3ab1e..1c19954e 100755 --- a/management/backup.py +++ b/management/backup.py @@ -498,7 +498,7 @@ def list_target_files(config): msg = f"Connection to rsync host failed: {reason}" raise ValueError(msg) - elif target.scheme == "s3": + if target.scheme == "s3": import boto3.s3 from botocore.exceptions import ClientError From bf27ac07ed71535278dafedb2047a9f11216a72f Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:37:36 -0700 Subject: [PATCH 34/37] Fixed EM102 (f-string-in-exception): Exception must not use an f-string literal, assign to variable first --- management/dns_update.py | 15 ++++++++++----- management/mail_log.py | 3 ++- management/mailconfig.py | 3 ++- tests/fail2ban.py | 3 ++- 4 files changed, 16 insertions(+), 8 deletions(-) diff --git a/management/dns_update.py b/management/dns_update.py index 840b98ad..b874624e 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -590,7 +590,8 @@ def get_dns_zonefile(zone, env): if zone == domain: break else: - raise ValueError(f"{zone} is not a domain name that corresponds to a zone.") + msg = f"{zone} is not a domain name that corresponds to a zone." + raise ValueError(msg) nsd_zonefile = "/etc/nsd/zones/" + fn with open(nsd_zonefile, encoding="utf-8") as f: @@ -896,7 +897,8 @@ def set_custom_dns_record(qname, rtype, value, action, env): else: # No match. if qname != "_secondary_nameserver": - raise ValueError(f"{qname} is not a domain name or a subdomain of a domain name managed by this box.") + msg = f"{qname} is not a domain name or a subdomain of a domain name managed by this box." + raise ValueError(msg) # validate rtype rtype = rtype.upper() @@ -926,7 +928,8 @@ def set_custom_dns_record(qname, rtype, value, action, env): # anything goes pass else: - raise ValueError(f"Unknown record type '{rtype}'.") + msg = f"Unknown record type '{rtype}'." + raise ValueError(msg) # load existing config config = list(get_custom_dns_config(env)) @@ -1037,7 +1040,8 @@ 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(f"Could not resolve the IP address of {item}.") + msg = f"Could not resolve the IP address of {item}." + raise ValueError(msg) else: # Validate IP address. try: @@ -1046,7 +1050,8 @@ def set_secondary_dns(hostnames, env): else: ipaddress.ip_address(item[4:]) # raises a ValueError if there's a problem except ValueError: - raise ValueError(f"'{item[4:]}' is not an IPv4 or IPv6 address or subnet.") + msg = f"'{item[4:]}' is not an IPv4 or IPv6 address or subnet." + raise ValueError(msg) # Set. set_custom_dns_record("_secondary_nameserver", "A", " ".join(hostnames), "set", env) diff --git a/management/mail_log.py b/management/mail_log.py index 60e5e649..8907342e 100755 --- a/management/mail_log.py +++ b/management/mail_log.py @@ -608,7 +608,8 @@ def valid_date(string): try: date = dateutil.parser.parse(string) except ValueError: - raise argparse.ArgumentTypeError(f"Unrecognized date and/or time '{string}'") + msg = f"Unrecognized date and/or time '{string}'" + raise argparse.ArgumentTypeError(msg) return date diff --git a/management/mailconfig.py b/management/mailconfig.py index 22c18fff..d9f38ede 100755 --- a/management/mailconfig.py +++ b/management/mailconfig.py @@ -448,7 +448,8 @@ def get_mail_password(email, env): c.execute('SELECT password FROM users WHERE email=?', (email,)) rows = c.fetchall() if len(rows) != 1: - raise ValueError(f"That's not a user ({email}).") + msg = f"That's not a user ({email})." + raise ValueError(msg) return rows[0][0] def remove_mail_user(email, env): diff --git a/tests/fail2ban.py b/tests/fail2ban.py index 30e3c267..46ce9271 100644 --- a/tests/fail2ban.py +++ b/tests/fail2ban.py @@ -142,7 +142,8 @@ def http_test(url, expected_status, postdata=None, qsargs=None, auth=None): # return response status code if r.status_code != expected_status: r.raise_for_status() # anything but 200 - raise OSError(f"Got unexpected status code {r.status_code}.") + msg = f"Got unexpected status code {r.status_code}." + raise OSError(msg) # define how to run a test From d15170b18cf6eee1a56c16a9bd97a09ab2ade677 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 04:40:22 -0700 Subject: [PATCH 35/37] Fixed RUF010 (explicit-f-string-type-conversion): Use explicit conversion flag --- management/ssl_certificates.py | 2 +- management/status_checks.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/management/ssl_certificates.py b/management/ssl_certificates.py index 5b9470ec..db36234c 100755 --- a/management/ssl_certificates.py +++ b/management/ssl_certificates.py @@ -515,7 +515,7 @@ def check_certificate(domain, ssl_certificate, ssl_private_key, warn_if_expiring cert = load_pem(ssl_cert_chain[0]) if not isinstance(cert, Certificate): raise ValueError("This is not a certificate file.") except ValueError as e: - return (f"There is a problem with the certificate file: {str(e)}", None) + return (f"There is a problem with the certificate file: {e!s}", None) # First check that the domain name is one of the names allowed by # the certificate. diff --git a/management/status_checks.py b/management/status_checks.py index b584f9e1..bfc820df 100755 --- a/management/status_checks.py +++ b/management/status_checks.py @@ -248,7 +248,7 @@ def check_free_disk_space(rounded_values, env, output): def check_free_memory(rounded_values, env, output): # Check free memory. percent_free = 100 - psutil.virtual_memory().percent - memory_msg = f"System memory is {str(round(percent_free))}% free." + memory_msg = f"System memory is {round(percent_free)!s}% free." if percent_free >= 20: if rounded_values: memory_msg = "System free memory is at least 20%." output.print_ok(memory_msg) From a568c6ff74d5d67c087d38940361e767b02e2c83 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 05:10:13 -0700 Subject: [PATCH 36/37] Fixed RET505 (superfluous-else-return): Unnecessary `elif` after `return` statement --- management/backup.py | 5 ++--- 1 file changed, 2 insertions(+), 3 deletions(-) diff --git a/management/backup.py b/management/backup.py index 1c19954e..d8807a54 100755 --- a/management/backup.py +++ b/management/backup.py @@ -528,7 +528,7 @@ def list_target_files(config): except ClientError as e: raise ValueError(e) return backup_list - elif target.scheme == 'b2': + if target.scheme == 'b2': from b2sdk.v1 import InMemoryAccountInfo, B2Api from b2sdk.v1.exception import NonExistentBucket info = InMemoryAccountInfo() @@ -547,8 +547,7 @@ def list_target_files(config): raise ValueError(msg) return [(key.file_name, key.size) for key, _ in bucket.ls()] - else: - raise ValueError(config["target"]) + raise ValueError(config["target"]) def backup_set_custom(env, target, target_user, target_pass, min_age): From 00280123abe6a4bc3c37ec33db9c7b74397c52c3 Mon Sep 17 00:00:00 2001 From: Teal Dulcet Date: Wed, 18 Jun 2025 05:14:12 -0700 Subject: [PATCH 37/37] Fixed RUF005 (collection-literal-concatenation): Consider iterable unpacking instead of concatenation --- management/dns_update.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/management/dns_update.py b/management/dns_update.py index b874624e..bf54062e 100755 --- a/management/dns_update.py +++ b/management/dns_update.py @@ -716,9 +716,9 @@ def sign_zone(domain, zonefile, env): # zonefile to sign "/etc/nsd/zones/" + zonefile, - ] # keys to sign with (order doesn't matter -- it'll figure it out) - + all_keys + *all_keys + ] ) # Create a DS record based on the patched-up key files. The DS record is specific to the