amorfo77/parsedmarc
1
0
Fork 0
mirror of https://github.com/domainaware/parsedmarc.git synced 2026-07-31 04:35:58 +00:00
Code Issues Projects Releases Wiki Activity

Upgrade ruff to 0.16.0 and pyright to 1.1.411; convert to f-strings (#847)

Browse Source 
* Upgrade ruff to 0.16.0 and pyright to 1.1.411; convert to f-strings

ruff 0.16.0 expanded the default lint rule selection well beyond the
long-standing E4/E7/E9/F, so [tool.ruff.lint] now selects the rule set
explicitly: the pre-0.16 defaults plus the modern-type-hint UP rules and
the two f-string rules (UP030/UP032). All 352 UP030/UP032 findings were
auto-fixed; conversions requiring Python 3.12 f-string quote reuse were
conservatively left as .format() by ruff (verified: the whole package and
test suite byte-compile under CPython 3.10.20, the oldest CI version).
Adopting the other newly-default rule families (BLE, SIM, C4, DTZ, I, ...)
is deferred as a deliberate per-family decision.

ruff format with 0.16.0 also now formats Python code fences in Markdown,
which reformatted one block in parsedmarc/resources/maps/AGENTS.md.

ruff check, ruff format --check, pyright (0 errors/warnings), and the
full test suite (775 passed) are green on the new versions.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

* Address Copilot review findings on the f-string conversion

- Rewrite messages that used backslash line continuations inside string
  literals, which embedded the source indentation as literal whitespace
  in the logged/raised text: the duplicate search-error messages in the
  Elasticsearch and OpenSearch outputs, the 'since'-option warning (which
  also implicitly concatenated "24hrs" and "SMTP" with no separator) and
  the IMAP 'since' debug line, and the missing-org_name KeyError message.
- Build the Splunk HEC newline-delimited payloads by appending to a list
  and joining once instead of quadratic string concatenation in a loop.

Declined: switching the webhook output's logger.error to
logger.exception — the single-line ERROR without a traceback is the
deliberate house pattern for batch-resilient sinks, and changing log
verbosity is out of scope for this refactor PR.

ruff check/format, pyright (0 errors/warnings), 775 tests, and a
CPython 3.10 compileall all pass.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Sean Whalen
2026-07-25 13:46:36 -04:00
committed by GitHub
co-authored by Claude Fable 5
parent df086e7242
commit 0a95a0ceb2
15 changed files with 242 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
CHANGELOG.md
+4
View File
@@ -8,6 +8,10 @@
- On OpenSearch Dashboards/Kibana, the table is now a TSVB visualization using a Filter Ratio metric (passed messages over total messages per `header_from`), since the previous agg-based data table can't compute a per-domain ratio. Editing the imported visualization on Kibana 8.x requires first enabling the `metrics:allowStringIndices` advanced setting.
- **Directory paths are now accepted as `file_path` CLI arguments** ([#397](https://github.com/domainaware/parsedmarc/issues/397)): a directory expands to the report files inside it using shell-glob semantics (dotfile entries excluded, subdirectories skipped by default), and the new `-r`/`--recursive` flag descends into subdirectories and enables `**` recursion in glob patterns.
### Changes
- **Updated the pinned dev-tooling versions and converted the codebase to f-strings**: `ruff` 0.15.21 → 0.16.0 and `pyright` 1.1.410 → 1.1.411 in the `[build]` extra. All `str.format()` calls were converted to f-strings (ruff rules `UP030`/`UP032`), except where the f-string form would require Python 3.12's quote reuse inside expressions — those keep `.format()` with the redundant positional indices removed. Because ruff 0.16.0 greatly expanded its default lint rule set (adding rule families such as `BLE`, `SIM`, `C4`, `DTZ`, and import sorting, some of which conflict with deliberate house style — e.g. `BLE001` flags the parser's intentional broad catches), `[tool.ruff.lint]` now selects its rule set explicitly: the pre-0.16 defaults (`E4`/`E7`/`E9`/`F`) plus the modern-type-hint and f-string `UP` rules. Adopting any of the newly-default rule families is left as a deliberate future per-family decision. Alongside the conversion, a few pre-existing string defects the conversion surfaced were cleaned up: log/exception messages that embedded runs of indentation whitespace via backslash line continuations (and one missing sentence separator in the `since`-option warning) now read cleanly, and the Splunk HEC output builds its newline-delimited payload by joining a list instead of repeated string concatenation. No functional behavior changes.
### Bug fixes
- **The Elasticsearch/OpenSearch aggregate dashboards' over-time charts (and the Grafana ES dashboard's summary pies and time series) bucketed on the multi-valued `date_range` field**; a date histogram counts a report once per value, double-counting any report whose begin and end dates fall in different buckets. All date histograms and time-range filters now use the single-valued `date_begin`, matching the report-begin semantics of the PostgreSQL (`begin_date`) and Splunk (`_time` = interval begin) dashboards.
parsedmarc/__init__.py
+73 -121
View File
@@ -67,7 +67,7 @@ from parsedmarc.utils import (
timestamp_to_human,
)
logger.debug("parsedmarc v{0}".format(__version__))
logger.debug(f"parsedmarc v{__version__}")
feedback_report_regex = re.compile(r"^([\w\-]+): (.+)$", re.MULTILINE)
xml_header_regex = re.compile(r"^<\?xml .*?>", re.MULTILINE)
@@ -159,7 +159,7 @@ def _exc_origin(error: BaseException) -> str:
if not frames:
return ""
last = frames[-1]
return " (raised at {0}:{1})".format(last.filename, last.lineno)
return f" (raised at {last.filename}:{last.lineno})"
def _text(value: Any) -> str | None:
@@ -824,7 +824,7 @@ def parse_aggregate_report_xml(
try:
xmltodict.parse(xml)["feedback"]
except Exception as e:
errors.append("Invalid XML: {0}".format(e.__str__()))
errors.append(f"Invalid XML: {e.__str__()}")
try:
tree = etree.parse(
BytesIO(xml.encode("utf-8")),
@@ -872,13 +872,10 @@ def parse_aggregate_report_xml(
if new_org_name is not None:
org_name = new_org_name
if not org_name:
logger.debug(
"Could not parse org_name from XML.\r\n{0}".format(report.__str__())
)
logger.debug(f"Could not parse org_name from XML.\r\n{report.__str__()}")
raise KeyError(
"Organization name is missing. \
This field is a requirement for \
saving the report"
"Organization name is missing. This field is a requirement "
"for saving the report"
)
new_report_metadata["org_name"] = org_name
new_report_metadata["org_email"] = report_metadata["email"]
@@ -977,14 +974,14 @@ def parse_aggregate_report_xml(
if policy_published["np"] is not None:
np_ = policy_published["np"]
if np_ not in ("none", "quarantine", "reject"):
logger.warning("Invalid np value: {0}".format(np_))
logger.warning(f"Invalid np value: {np_}")
new_policy_published["np"] = np_
testing = None
if "testing" in policy_published:
if policy_published["testing"] is not None:
testing = policy_published["testing"]
if testing not in ("n", "y"):
logger.warning("Invalid testing value: {0}".format(testing))
logger.warning(f"Invalid testing value: {testing}")
new_policy_published["testing"] = testing
discovery_method = None
if "discovery_method" in policy_published:
@@ -992,7 +989,7 @@ def parse_aggregate_report_xml(
discovery_method = policy_published["discovery_method"]
if discovery_method not in ("psl", "treewalk"):
logger.warning(
"Invalid discovery_method value: {0}".format(discovery_method)
f"Invalid discovery_method value: {discovery_method}"
)
new_policy_published["discovery_method"] = discovery_method
new_report["policy_published"] = new_policy_published
@@ -1002,7 +999,7 @@ def parse_aggregate_report_xml(
if keep_alive is not None and i > 0 and i % 20 == 0:
logger.debug("Sending keepalive cmd")
keep_alive()
logger.debug("Processed {0}/{1}".format(i, len(report["record"])))
logger.debug("Processed {}/{}".format(i, len(report["record"])))
try:
report_record = _parse_report_record(
report["record"][i],
@@ -1024,7 +1021,7 @@ def parse_aggregate_report_xml(
normalize=normalize_timespan,
)
except Exception as e:
logger.warning("Could not parse record: {0}".format(e))
logger.warning(f"Could not parse record: {e}")
else:
report_record = _parse_report_record(
@@ -1052,20 +1049,16 @@ def parse_aggregate_report_xml(
return cast(AggregateReport, new_report)
except expat.ExpatError as error:
raise InvalidAggregateReport(
"Invalid XML: {0}".format(error.__str__())
) from error
raise InvalidAggregateReport(f"Invalid XML: {error.__str__()}") from error
except KeyError as error:
raise InvalidAggregateReport(
"Missing field: {0}".format(error.__str__())
) from error
raise InvalidAggregateReport(f"Missing field: {error.__str__()}") from error
except AttributeError as error:
raise InvalidAggregateReport("Report missing required section") from error
except Exception as error:
raise InvalidAggregateReport(
"Unexpected error: {0}{1}".format(error.__str__(), _exc_origin(error))
f"Unexpected error: {error.__str__()}{_exc_origin(error)}"
) from error
@@ -1143,7 +1136,7 @@ def extract_report(content: bytes | str | BinaryIO) -> str:
except Exception as error:
raise ParserError(
"Invalid archive file: {0}{1}".format(error.__str__(), _exc_origin(error))
f"Invalid archive file: {error.__str__()}{_exc_origin(error)}"
) from error
finally:
if file_object:
@@ -1607,13 +1600,11 @@ def parse_failure_report(
return cast(FailureReport, parsed_report)
except KeyError as error:
raise InvalidFailureReport(
"Missing value: {0}".format(error.__str__())
) from error
raise InvalidFailureReport(f"Missing value: {error.__str__()}") from error
except Exception as error:
raise InvalidFailureReport(
"Unexpected error: {0}{1}".format(error.__str__(), _exc_origin(error))
f"Unexpected error: {error.__str__()}{_exc_origin(error)}"
) from error
@@ -1789,7 +1780,7 @@ def parse_report_email(
sample = None
is_feedback_report: bool = False
if "From" in msg_headers:
logger.info("Parsing mail from {0} on {1}".format(msg_headers["From"], date))
logger.info("Parsing mail from {} on {}".format(msg_headers["From"], date))
if "Subject" in msg_headers:
subject = msg_headers["Subject"]
for part in msg.walk():
@@ -1839,9 +1830,7 @@ def parse_report_email(
fields["received-date"], fields["sender-ip-address"]
)
except Exception as e:
error = 'Unable to parse message with subject "{0}": {1}{2}'.format(
subject, e, _exc_origin(e)
)
error = f'Unable to parse message with subject "{subject}": {e}{_exc_origin(e)}'
raise InvalidDMARCReport(error) from e
sample = parts[1].lstrip()
@@ -1883,15 +1872,12 @@ def parse_report_email(
except InvalidDMARCReport as e:
error = (
'Message with subject "{0}" is not a valid '
"DMARC report: {1}".format(subject, e)
f'Message with subject "{subject}" is not a valid DMARC report: {e}'
)
raise ParserError(error) from e
except Exception as e:
error = 'Unable to parse message with subject "{0}": {1}{2}'.format(
subject, e, _exc_origin(e)
)
error = f'Unable to parse message with subject "{subject}": {e}{_exc_origin(e)}'
raise ParserError(error) from e
if feedback_report and sample:
@@ -1912,9 +1898,9 @@ def parse_report_email(
)
except InvalidFailureReport as e:
error = (
'Message with subject "{0}" '
f'Message with subject "{subject}" '
"is not a valid "
"failure DMARC report: {1}".format(subject, e)
f"failure DMARC report: {e}"
)
raise InvalidFailureReport(error) from e
@@ -1922,7 +1908,7 @@ def parse_report_email(
return result
if result is None:
error = 'Message with subject "{0}" is not a valid report'.format(subject)
error = f'Message with subject "{subject}" is not a valid report'
raise InvalidDMARCReport(error)
return result
@@ -1965,11 +1951,11 @@ def _describe_parse_failure(
sniff = sniff.lstrip()
if sniff.startswith("<"):
return "Invalid aggregate report: {0}".format(aggregate_error)
return f"Invalid aggregate report: {aggregate_error}"
if sniff.startswith("{"):
return "Invalid SMTP TLS report: {0}".format(smtp_tls_error)
return f"Invalid SMTP TLS report: {smtp_tls_error}"
if _looks_like_email(sniff):
return "Invalid report email: {0}".format(email_error)
return f"Invalid report email: {email_error}"
return (
"Not a recognized report format (not a DMARC aggregate XML report, "
"an SMTP TLS JSON report, or a DMARC report email)"
@@ -2017,7 +2003,7 @@ def parse_report_file(
file_object: BinaryIO
if isinstance(input_, (str, os.PathLike)):
file_path = os.fspath(input_)
logger.debug("Parsing {0}".format(file_path))
logger.debug(f"Parsing {file_path}")
file_object = open(file_path, "rb")
elif isinstance(input_, (bytes, bytearray, memoryview)):
file_object = BytesIO(bytes(input_))
@@ -2126,10 +2112,10 @@ def get_dmarc_reports_from_mbox(
mbox = mailbox.mbox(input_)
message_keys = mbox.keys()
total_messages = len(message_keys)
logger.debug("Found {0} messages in {1}".format(total_messages, input_))
logger.debug(f"Found {total_messages} messages in {input_}")
for i in tqdm(range(total_messages), disable=None):
message_key = message_keys[i]
logger.info("Processing message {0} of {1}".format(i + 1, total_messages))
logger.info(f"Processing message {i + 1} of {total_messages}")
msg_content = mbox.get_string(message_key)
try:
sa = strip_attachment_payloads
@@ -2165,7 +2151,7 @@ def get_dmarc_reports_from_mbox(
except InvalidDMARCReport as error:
logger.warning(error.__str__())
except mailbox.NoSuchMailboxError:
raise InvalidDMARCReport("Mailbox {0} does not exist".format(input_))
raise InvalidDMARCReport(f"Mailbox {input_} does not exist")
return {
"aggregate_reports": aggregate_reports,
"failure_reports": failure_reports,
@@ -2191,8 +2177,8 @@ def _migrate_forensic_archive_folder(
(warn, don't crash). Uses the folder-management API added in mailsuite
2.1.0 (``folder_exists`` / ``rename_folder`` / ``merge_folders``).
"""
old_folder = "{0}/Forensic".format(archive_folder)
new_folder = "{0}/Failure".format(archive_folder)
old_folder = f"{archive_folder}/Forensic"
new_folder = f"{archive_folder}/Failure"
try:
if not connection.folder_exists(old_folder):
return
@@ -2201,23 +2187,13 @@ def _migrate_forensic_archive_folder(
# created Failure folder): move the legacy folder's messages into
# the new one and drop the now-empty legacy folder.
connection.merge_folders(old_folder, new_folder)
logger.info(
"Merged legacy archive folder {0} into {1}".format(
old_folder, new_folder
)
)
logger.info(f"Merged legacy archive folder {old_folder} into {new_folder}")
else:
connection.rename_folder(old_folder, new_folder)
logger.info(
"Renamed legacy archive folder {0} to {1}".format(
old_folder, new_folder
)
)
logger.info(f"Renamed legacy archive folder {old_folder} to {new_folder}")
except Exception as error:
logger.warning(
"Could not migrate legacy archive folder {0} to {1}: {2}".format(
old_folder, new_folder, error
)
f"Could not migrate legacy archive folder {old_folder} to {new_folder}: {error}"
)
@@ -2290,10 +2266,10 @@ def get_dmarc_reports_from_mailbox(
aggregate_report_msg_uids = []
failure_report_msg_uids = []
smtp_tls_msg_uids = []
aggregate_reports_folder = "{0}/Aggregate".format(archive_folder)
failure_reports_folder = "{0}/Failure".format(archive_folder)
smtp_tls_reports_folder = "{0}/SMTP-TLS".format(archive_folder)
invalid_reports_folder = "{0}/Invalid".format(archive_folder)
aggregate_reports_folder = f"{archive_folder}/Aggregate"
failure_reports_folder = f"{archive_folder}/Failure"
smtp_tls_reports_folder = f"{archive_folder}/SMTP-TLS"
invalid_reports_folder = f"{archive_folder}/Invalid"
if results:
aggregate_reports = results["aggregate_reports"].copy()
@@ -2322,17 +2298,17 @@ def get_dmarc_reports_from_mailbox(
_since = int(s[1]) * 60 * 24 * 7
else:
logger.warning(
"Incorrect format for 'since' option. \
Provided value:{0}, Expected values:(5m|3h|2d|1w). \
Ignoring option, fetching messages for last 24hrs"
"SMTP does not support a time or timezone in since."
"See https://www.rfc-editor.org/rfc/rfc3501#page-52".format(since)
f"Incorrect format for 'since' option. Provided value: {since}, "
"expected values: (5m|3h|2d|1w). Ignoring option, fetching "
"messages for last 24hrs. SMTP does not support a time or "
"timezone in since. See "
"https://www.rfc-editor.org/rfc/rfc3501#page-52"
)
if isinstance(connection, IMAPConnection):
logger.debug(
"Only days and weeks values in 'since' option are \
considered for IMAP connections. Examples: 2d or 1w"
"Only days and weeks values in 'since' option are considered "
"for IMAP connections. Examples: 2d or 1w"
)
since = (datetime.now(timezone.utc) - timedelta(minutes=_since)).strftime(
"%d-%b-%Y"
@@ -2353,22 +2329,18 @@ def get_dmarc_reports_from_mailbox(
reports_folder, batch_size=batch_size, since=since
)
total_messages = len(messages)
logger.debug("Found {0} messages in {1}".format(len(messages), reports_folder))
logger.debug(f"Found {len(messages)} messages in {reports_folder}")
if batch_size and not since:
message_limit = min(total_messages, batch_size)
else:
message_limit = total_messages
logger.debug("Processing {0} messages".format(message_limit))
logger.debug(f"Processing {message_limit} messages")
for i in range(message_limit):
msg_uid = messages[i]
logger.debug(
"Processing message {0} of {1}: UID {2}".format(
i + 1, message_limit, msg_uid
)
)
logger.debug(f"Processing message {i + 1} of {message_limit}: UID {msg_uid}")
message_id: int | str
if isinstance(connection, IMAPConnection):
message_id = int(msg_uid)
@@ -2420,16 +2392,14 @@ def get_dmarc_reports_from_mailbox(
logger.warning(error.__str__())
if not test:
if delete:
logger.debug("Deleting message UID {0}".format(msg_uid))
logger.debug(f"Deleting message UID {msg_uid}")
if isinstance(connection, IMAPConnection):
connection.delete_message(int(message_id))
else:
connection.delete_message(str(message_id))
else:
logger.debug(
"Moving message UID {0} to {1}".format(
msg_uid, invalid_reports_folder
)
f"Moving message UID {msg_uid} to {invalid_reports_folder}"
)
if isinstance(connection, IMAPConnection):
connection.move_message(int(message_id), invalid_reports_folder)
@@ -2446,81 +2416,63 @@ def get_dmarc_reports_from_mailbox(
for i in range(number_of_processed_msgs):
msg_uid = processed_messages[i]
logger.debug(
"Deleting message {0} of {1}: UID {2}".format(
i + 1, number_of_processed_msgs, msg_uid
)
f"Deleting message {i + 1} of {number_of_processed_msgs}: UID {msg_uid}"
)
try:
connection.delete_message(msg_uid)
except Exception as e:
message = "Error deleting message UID"
e = "{0} {1}: {2}".format(message, msg_uid, e)
logger.error("Mailbox error: {0}".format(e))
e = f"{message} {msg_uid}: {e}"
logger.error(f"Mailbox error: {e}")
else:
if len(aggregate_report_msg_uids) > 0:
log_message = "Moving aggregate report messages from"
logger.debug(
"{0} {1} to {2}".format(
log_message, reports_folder, aggregate_reports_folder
)
f"{log_message} {reports_folder} to {aggregate_reports_folder}"
)
number_of_agg_report_msgs = len(aggregate_report_msg_uids)
for i in range(number_of_agg_report_msgs):
msg_uid = aggregate_report_msg_uids[i]
logger.debug(
"Moving message {0} of {1}: UID {2}".format(
i + 1, number_of_agg_report_msgs, msg_uid
)
f"Moving message {i + 1} of {number_of_agg_report_msgs}: UID {msg_uid}"
)
try:
connection.move_message(msg_uid, aggregate_reports_folder)
except Exception as e:
message = "Error moving message UID"
e = "{0} {1}: {2}".format(message, msg_uid, e)
logger.error("Mailbox error: {0}".format(e))
e = f"{message} {msg_uid}: {e}"
logger.error(f"Mailbox error: {e}")
if len(failure_report_msg_uids) > 0:
message = "Moving failure report messages from"
logger.debug(
"{0} {1} to {2}".format(
message, reports_folder, failure_reports_folder
)
)
logger.debug(f"{message} {reports_folder} to {failure_reports_folder}")
number_of_failure_msgs = len(failure_report_msg_uids)
for i in range(number_of_failure_msgs):
msg_uid = failure_report_msg_uids[i]
message = "Moving message"
logger.debug(
"{0} {1} of {2}: UID {3}".format(
message, i + 1, number_of_failure_msgs, msg_uid
)
f"{message} {i + 1} of {number_of_failure_msgs}: UID {msg_uid}"
)
try:
connection.move_message(msg_uid, failure_reports_folder)
except Exception as e:
e = "Error moving message UID {0}: {1}".format(msg_uid, e)
logger.error("Mailbox error: {0}".format(e))
e = f"Error moving message UID {msg_uid}: {e}"
logger.error(f"Mailbox error: {e}")
if len(smtp_tls_msg_uids) > 0:
message = "Moving SMTP TLS report messages from"
logger.debug(
"{0} {1} to {2}".format(
message, reports_folder, smtp_tls_reports_folder
)
)
logger.debug(f"{message} {reports_folder} to {smtp_tls_reports_folder}")
number_of_smtp_tls_uids = len(smtp_tls_msg_uids)
for i in range(number_of_smtp_tls_uids):
msg_uid = smtp_tls_msg_uids[i]
message = "Moving message"
logger.debug(
"{0} {1} of {2}: UID {3}".format(
message, i + 1, number_of_smtp_tls_uids, msg_uid
)
f"{message} {i + 1} of {number_of_smtp_tls_uids}: UID {msg_uid}"
)
try:
connection.move_message(msg_uid, smtp_tls_reports_folder)
except Exception as e:
e = "Error moving message UID {0}: {1}".format(msg_uid, e)
logger.error("Mailbox error: {0}".format(e))
e = f"Error moving message UID {msg_uid}: {e}"
logger.error(f"Mailbox error: {e}")
results = {
"aggregate_reports": aggregate_reports,
"failure_reports": failure_reports,
@@ -2737,7 +2689,7 @@ def save_output(
if os.path.exists(output_directory):
if not os.path.isdir(output_directory):
raise ValueError("{0} is not a directory".format(output_directory))
raise ValueError(f"{output_directory} is not a directory")
else:
os.makedirs(output_directory)
@@ -2784,11 +2736,11 @@ def save_output(
while filename in sample_filenames:
message_count += 1
filename = "{0} ({1})".format(subject, message_count)
filename = f"{subject} ({message_count})"
sample_filenames.append(filename)
filename = "{0}.eml".format(filename)
filename = f"{filename}.eml"
path = os.path.join(samples_directory, filename)
with open(path, "w", newline="\n", encoding="utf-8") as sample_file:
sample_file.write(sample)
@@ -2858,12 +2810,12 @@ def _build_report_email_content(
attachment_filename += ".zip"
filename = attachment_filename
else:
filename = "DMARC-{0}.zip".format(date_string)
filename = f"DMARC-{date_string}.zip"
if subject is None:
subject = "DMARC results for {0}".format(date_string)
subject = f"DMARC results for {date_string}"
if message is None:
message = "DMARC results for {0}".format(date_string)
message = f"DMARC results for {date_string}"
zip_bytes = get_report_zip(results)
attachments = [(filename, zip_bytes)]
parsedmarc/cli.py
+33 -39
View File
@@ -100,7 +100,7 @@ def _normalize_graph_auth_method(value: str) -> str:
if method.name.lower() == value_lower:
return method.name
raise ConfigurationError(
"Invalid msgraph auth_method: {0!r}. Valid values are: {1}".format(
"Invalid msgraph auth_method: {!r}. Valid values are: {}".format(
value, ", ".join(m.name for m in AuthMethod)
)
)
@@ -125,12 +125,12 @@ def _msgraph_request_id_suffix(error: Exception) -> str:
request_id = headers.get("request-id")
parts = []
if request_id:
parts.append("request-id={0}".format(request_id))
parts.append(f"request-id={request_id}")
if client_request_id:
parts.append("client-request-id={0}".format(client_request_id))
parts.append(f"client-request-id={client_request_id}")
if not parts:
return ""
return " ({0})".format(", ".join(parts))
return " ({})".format(", ".join(parts))
except Exception:
return ""
@@ -151,11 +151,11 @@ def _log_msgraph_failure(
if isinstance(error, APIError):
detail = getattr(error, "primary_message", None) or error.message or str(error)
detail = " ".join(str(detail).split())
summary = "{0} status={1}: {2}".format(
type(error).__name__, error.response_status_code, detail
summary = (
f"{type(error).__name__} status={error.response_status_code}: {detail}"
)
else:
summary = "{0}: {1}".format(type(error).__name__, " ".join(str(error).split()))
summary = "{}: {}".format(type(error).__name__, " ".join(str(error).split()))
logger.error(
"Microsoft Graph %s failed (mailbox=%s, tenant_id=%s, auth_method=%s): %s%s",
@@ -302,9 +302,7 @@ def _read_secret_file(env_key: str, raw_path: str) -> str:
return f.read().rstrip("\r\n")
except (OSError, UnicodeDecodeError) as exc:
raise ConfigurationError(
"Cannot read secret file for {0}: {1} ({2})".format(
env_key, path, exc.__class__.__name__
)
f"Cannot read secret file for {env_key}: {path} ({exc.__class__.__name__})"
) from exc
@@ -401,7 +399,7 @@ def _configure_logging(log_level, log_file=None):
fh.setFormatter(formatter)
logger.addHandler(fh)
except (IOError, OSError, PermissionError) as error:
logger.warning("Unable to write to log file: {}".format(error))
logger.warning(f"Unable to write to log file: {error}")
# Loggers of the libraries that implement the mailbox and Microsoft Graph
@@ -524,10 +522,10 @@ def _load_config(config_file: str | None = None) -> ConfigParser:
if config_file is not None:
abs_path = os.path.abspath(config_file)
if not os.path.exists(abs_path):
raise ConfigurationError("A file does not exist at {0}".format(abs_path))
raise ConfigurationError(f"A file does not exist at {abs_path}")
if not os.access(abs_path, os.R_OK):
raise ConfigurationError(
"Unable to read {0} — check file permissions".format(abs_path)
f"Unable to read {abs_path} — check file permissions"
)
config.read(config_file)
_apply_env_overrides(config)
@@ -606,13 +604,11 @@ def _parse_config(config: ConfigParser, opts):
)
except Exception as ns_error:
raise ConfigurationError(
"DNS pre-flight check failed: {}".format(ns_error)
f"DNS pre-flight check failed: {ns_error}"
) from ns_error
if not dummy_hostname:
raise ConfigurationError(
"DNS pre-flight check failed: no PTR record for {} from {}".format(
opts.dns_test_address, opts.nameservers
)
f"DNS pre-flight check failed: no PTR record for {opts.dns_test_address} from {opts.nameservers}"
)
if "save_aggregate" in general_config:
opts.save_aggregate = bool(general_config.getboolean("save_aggregate"))
@@ -1445,14 +1441,14 @@ def _init_output_clients(opts):
es_smtp_tls_index = "smtp_tls"
if opts.elasticsearch_index_suffix:
suffix = opts.elasticsearch_index_suffix
es_aggregate_index = "{0}_{1}".format(es_aggregate_index, suffix)
es_failure_index = "{0}_{1}".format(es_failure_index, suffix)
es_smtp_tls_index = "{0}_{1}".format(es_smtp_tls_index, suffix)
es_aggregate_index = f"{es_aggregate_index}_{suffix}"
es_failure_index = f"{es_failure_index}_{suffix}"
es_smtp_tls_index = f"{es_smtp_tls_index}_{suffix}"
if opts.elasticsearch_index_prefix:
prefix = opts.elasticsearch_index_prefix
es_aggregate_index = "{0}{1}".format(prefix, es_aggregate_index)
es_failure_index = "{0}{1}".format(prefix, es_failure_index)
es_smtp_tls_index = "{0}{1}".format(prefix, es_smtp_tls_index)
es_aggregate_index = f"{prefix}{es_aggregate_index}"
es_failure_index = f"{prefix}{es_failure_index}"
es_smtp_tls_index = f"{prefix}{es_smtp_tls_index}"
elastic_timeout_value = (
float(opts.elasticsearch_timeout)
if opts.elasticsearch_timeout is not None
@@ -1490,14 +1486,14 @@ def _init_output_clients(opts):
os_smtp_tls_index = "smtp_tls"
if opts.opensearch_index_suffix:
suffix = opts.opensearch_index_suffix
os_aggregate_index = "{0}_{1}".format(os_aggregate_index, suffix)
os_failure_index = "{0}_{1}".format(os_failure_index, suffix)
os_smtp_tls_index = "{0}_{1}".format(os_smtp_tls_index, suffix)
os_aggregate_index = f"{os_aggregate_index}_{suffix}"
os_failure_index = f"{os_failure_index}_{suffix}"
os_smtp_tls_index = f"{os_smtp_tls_index}_{suffix}"
if opts.opensearch_index_prefix:
prefix = opts.opensearch_index_prefix
os_aggregate_index = "{0}{1}".format(prefix, os_aggregate_index)
os_failure_index = "{0}{1}".format(prefix, os_failure_index)
os_smtp_tls_index = "{0}{1}".format(prefix, os_smtp_tls_index)
os_aggregate_index = f"{prefix}{os_aggregate_index}"
os_failure_index = f"{prefix}{os_failure_index}"
os_smtp_tls_index = f"{prefix}{os_smtp_tls_index}"
opensearch_timeout_value = (
float(opts.opensearch_timeout)
if opts.opensearch_timeout is not None
@@ -1603,8 +1599,8 @@ def _main():
reports_["smtp_tls_reports"] = filtered_tls
indent_value = 2 if opts.prettify_json else None
output_str = "{0}\n".format(
json.dumps(reports_, ensure_ascii=False, indent=indent_value)
output_str = (
f"{json.dumps(reports_, ensure_ascii=False, indent=indent_value)}\n"
)
if not opts.silent:
@@ -1937,7 +1933,7 @@ def _main():
if opts.fail_on_output_error and output_errors:
raise ParserError(
"Output destination failures detected: {0}".format(
"Output destination failures detected: {}".format(
" | ".join(output_errors)
)
)
@@ -2254,7 +2250,7 @@ def _main():
fh.setFormatter(formatter)
logger.addHandler(fh)
except Exception as error:
logger.warning("Unable to write to log file: {}".format(error))
logger.warning(f"Unable to write to log file: {error}")
opts.active_log_file = opts.log_file
_configure_dependency_logging(logger.level)
@@ -2315,7 +2311,7 @@ def _main():
time.sleep(retry_delay)
retry_delay *= 2
else:
logger.error("Output client error: {0}".format(error_))
logger.error(f"Output client error: {error_}")
exit(1)
# Always close output clients on the way out (normal return,
@@ -2447,7 +2443,7 @@ def _main():
for result in results:
if isinstance(result[0], ParserError) or result[0] is None:
logger.error("Failed to parse {0} - {1}".format(result[1], result[0]))
logger.error(f"Failed to parse {result[1]} - {result[0]}")
else:
if result[0]["report_type"] == "aggregate":
report_org = result[0]["report"]["report_metadata"]["org_name"]
@@ -2809,7 +2805,7 @@ def _main():
config_reloading=lambda: _reload_requested or _shutdown_requested,
)
except FileExistsError as error:
logger.error("{0}".format(error.__str__()))
logger.error(f"{error.__str__()}")
exit(1)
except ParserError as error:
logger.error(error.__str__())
@@ -2939,9 +2935,7 @@ def _main():
fh.setFormatter(file_formatter)
logger.addHandler(fh)
except Exception as log_error:
logger.warning(
"Unable to write to log file: {}".format(log_error)
)
logger.warning(f"Unable to write to log file: {log_error}")
opts.active_log_file = new_log_file
_configure_dependency_logging(logger.level)
parsedmarc/elastic.py
+33 -43
View File
@@ -361,9 +361,7 @@ class _AggregateReportDoc(Document):
human_result=human_result,
)
)
self.dkim_results_combined.append(
"{0} / {1} / {2}".format(selector, domain, result)
)
self.dkim_results_combined.append(f"{selector} / {domain} / {result}")
def add_spf_result(
self,
@@ -380,9 +378,7 @@ class _AggregateReportDoc(Document):
human_result=human_result,
)
)
self.spf_results_combined.append(
"{0} / {1} / {2}".format(scope, domain, result)
)
self.spf_results_combined.append(f"{scope} / {domain} / {result}")
def save(self, **kwargs): # pyright: ignore[reportIncompatibleMethodOverride]
self.passed_dmarc = False
@@ -620,9 +616,7 @@ def set_hosts(
if not isinstance(hosts, list):
hosts = [hosts]
scheme = "https://" if use_ssl else "http://"
normalized_hosts = [
host if "://" in host else "{0}{1}".format(scheme, host) for host in hosts
]
normalized_hosts = [host if "://" in host else f"{scheme}{host}" for host in hosts]
conn_params = {"hosts": normalized_hosts, "request_timeout": timeout}
if use_ssl:
if ssl_cert_path:
@@ -675,12 +669,12 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
# — it is never installed as a mapping. See issue #169 and the
# *_combined fields on _AggregateReportDoc.
if not index.exists():
logger.debug("Creating Elasticsearch index: {0}".format(name))
logger.debug(f"Creating Elasticsearch index: {name}")
if effective_settings:
index.settings(**effective_settings)
index.create()
except Exception as e:
raise ElasticsearchError("Elasticsearch error: {0}".format(e.__str__()))
raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
def migrate_indexes(
@@ -843,11 +837,11 @@ def save_aggregate_report_to_elasticsearch(
end_date_query = Q(dict(range=dict(date_end=dict(lte=end_date)))) # pyright: ignore[reportArgumentType]
if index_suffix is not None:
search_index = "dmarc_aggregate_{0}*".format(index_suffix)
search_index = f"dmarc_aggregate_{index_suffix}*"
else:
search_index = "dmarc_aggregate*"
if index_prefix is not None:
search_index = "{0}{1}".format(index_prefix, search_index)
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query & domain_query
query = query & begin_date_query & end_date_query
@@ -862,18 +856,15 @@ def save_aggregate_report_to_elasticsearch(
existing = search.execute()
except Exception as error_:
raise ElasticsearchError(
"Elasticsearch's search for existing report \
error: {}".format(error_.__str__())
f"Elasticsearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
raise AlreadySaved(
"An aggregate report ID {0} from {1} about {2} "
"with a date range of {3} UTC to {4} UTC already "
f"An aggregate report ID {report_id} from {org_name} about {domain} "
f"with a date range of {begin_date_human} UTC to {end_date_human} UTC already "
"exists in "
"Elasticsearch".format(
report_id, org_name, domain, begin_date_human, end_date_human
)
"Elasticsearch"
)
published_policy = _PublishedPolicy(
domain=aggregate_report["policy_published"]["domain"],
@@ -966,11 +957,11 @@ def save_aggregate_report_to_elasticsearch(
index = "dmarc_aggregate"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = f"{index_prefix}{index}"
index = "{0}-{1}".format(index, index_date)
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -980,7 +971,7 @@ def save_aggregate_report_to_elasticsearch(
try:
agg_doc.save()
except Exception as e:
raise ElasticsearchError("Elasticsearch error: {0}".format(e.__str__()))
raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
def save_failure_report_to_elasticsearch(
@@ -1028,12 +1019,12 @@ def save_failure_report_to_elasticsearch(
arrival_date_epoch_milliseconds = int(arrival_date.timestamp() * 1000)
if index_suffix is not None:
search_index = "dmarc_failure_{0}*,dmarc_forensic_{0}*".format(index_suffix)
search_index = f"dmarc_failure_{index_suffix}*,dmarc_forensic_{index_suffix}*"
else:
search_index = "dmarc_failure*,dmarc_forensic*"
if index_prefix is not None:
search_index = ",".join(
"{0}{1}".format(index_prefix, part) for part in search_index.split(",")
f"{index_prefix}{part}" for part in search_index.split(",")
)
search = Search(index=search_index)
q = Q(dict(match=dict(arrival_date=arrival_date_epoch_milliseconds))) # pyright: ignore[reportArgumentType]
@@ -1085,8 +1076,8 @@ def save_failure_report_to_elasticsearch(
if len(existing) > 0:
raise AlreadySaved(
"A failure sample to {0} from {1} "
"with a subject of {2} and arrival date of {3} "
"A failure sample to {} from {} "
"with a subject of {} and arrival date of {} "
"already exists in "
"Elasticsearch".format(
to_, from_, subject, failure_report["arrival_date_utc"]
@@ -1147,14 +1138,14 @@ def save_failure_report_to_elasticsearch(
index = "dmarc_failure"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = f"{index_prefix}{index}"
if monthly_indexes:
index_date = arrival_date.strftime("%Y-%m")
else:
index_date = arrival_date.strftime("%Y-%m-%d")
index = "{0}-{1}".format(index, index_date)
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -1163,10 +1154,10 @@ def save_failure_report_to_elasticsearch(
try:
failure_doc.save()
except Exception as e:
raise ElasticsearchError("Elasticsearch error: {0}".format(e.__str__()))
raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
except KeyError as e:
raise InvalidFailureReport(
"Failure report missing required field: {0}".format(e.__str__())
f"Failure report missing required field: {e.__str__()}"
)
@@ -1213,11 +1204,11 @@ def save_smtp_tls_report_to_elasticsearch(
end_date_query = Q(dict(match=dict(date_end=end_date))) # pyright: ignore[reportArgumentType]
if index_suffix is not None:
search_index = "smtp_tls_{0}*".format(index_suffix)
search_index = f"smtp_tls_{index_suffix}*"
else:
search_index = "smtp_tls*"
if index_prefix is not None:
search_index = "{0}{1}".format(index_prefix, search_index)
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query
query = query & begin_date_query & end_date_query
@@ -1227,8 +1218,7 @@ def save_smtp_tls_report_to_elasticsearch(
existing = search.execute()
except Exception as error_:
raise ElasticsearchError(
"Elasticsearch's search for existing report \
error: {}".format(error_.__str__())
f"Elasticsearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
@@ -1242,10 +1232,10 @@ def save_smtp_tls_report_to_elasticsearch(
index = "smtp_tls"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = "{0}-{1}".format(index, index_date)
index = f"{index_prefix}{index}"
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -1274,7 +1264,7 @@ def save_smtp_tls_report_to_elasticsearch(
policy_domain_combined = policy.get("policy_domain") or "none"
policy_type_combined = policy.get("policy_type") or "none"
smtp_tls_doc.policies_combined.append(
"{0} / {1}".format(policy_domain_combined, policy_type_combined)
f"{policy_domain_combined} / {policy_type_combined}"
)
policy_doc = _SMTPTLSPolicyDoc(
policy_domain=policy["policy_domain"],
@@ -1327,7 +1317,7 @@ def save_smtp_tls_report_to_elasticsearch(
failure_reason_code=failure_reason_code,
)
smtp_tls_doc.failure_details_combined.append(
"{0} / {1} / {2} / {3} / {4} / {5}".format(
"{} / {} / {} / {} / {} / {}".format(
policy_domain_combined,
policy_type_combined,
failure_detail.get("result_type") or "none",
@@ -1344,7 +1334,7 @@ def save_smtp_tls_report_to_elasticsearch(
try:
smtp_tls_doc.save()
except Exception as e:
raise ElasticsearchError("Elasticsearch error: {0}".format(e.__str__()))
raise ElasticsearchError(f"Elasticsearch error: {e.__str__()}")
# Backward-compatible aliases
parsedmarc/kafkaclient.py
+8 -8
View File
@@ -60,7 +60,7 @@ class KafkaClient(object):
config: dict[str, Any] = dict(
value_serializer=lambda v: json.dumps(v).encode("utf-8"),
bootstrap_servers=kafka_hosts,
client_id="parsedmarc-{0}".format(__version__),
client_id=f"parsedmarc-{__version__}",
)
if ssl or username or password:
config["security_protocol"] = "SSL"
@@ -106,7 +106,7 @@ class KafkaClient(object):
begin_date_human = begin_date.strftime("%Y-%m-%dT%H:%M:%S")
end_date_human = end_date.strftime("%Y-%m-%dT%H:%M:%S")
date_range = [begin_date_human, end_date_human]
logger.debug("date_range is {}".format(date_range))
logger.debug(f"date_range is {date_range}")
return date_range
def save_aggregate_reports_to_kafka(
@@ -148,11 +148,11 @@ class KafkaClient(object):
"Kafka error: Unknown topic or partition on broker"
)
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
try:
self.producer.flush()
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
def save_failure_reports_to_kafka(
self,
@@ -182,11 +182,11 @@ class KafkaClient(object):
except UnknownTopicOrPartitionError:
raise KafkaError("Kafka error: Unknown topic or partition on broker")
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
try:
self.producer.flush()
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
# Backward-compatible alias
save_forensic_reports_to_kafka = save_failure_reports_to_kafka
@@ -219,8 +219,8 @@ class KafkaClient(object):
except UnknownTopicOrPartitionError:
raise KafkaError("Kafka error: Unknown topic or partition on broker")
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
try:
self.producer.flush()
except Exception as e:
raise KafkaError("Kafka error: {0}".format(e.__str__()))
raise KafkaError(f"Kafka error: {e.__str__()}")
parsedmarc/loganalytics.py
+1 -1
View File
@@ -129,7 +129,7 @@ class LogAnalyticsClient(object):
try:
logs_client.upload(self.conf.dcr_immutable_id, dcr_stream, results)
except HttpResponseError as e:
raise LogAnalyticsException("Upload failed: {error}".format(error=e))
raise LogAnalyticsException(f"Upload failed: {e}")
def publish_results(
self,
parsedmarc/opensearch.py
+33 -41
View File
@@ -319,9 +319,7 @@ class _AggregateReportDoc(Document):
human_result=human_result,
)
)
self.dkim_results_combined.append(
"{0} / {1} / {2}".format(selector, domain, result)
)
self.dkim_results_combined.append(f"{selector} / {domain} / {result}")
def add_spf_result(
self,
@@ -338,9 +336,7 @@ class _AggregateReportDoc(Document):
human_result=human_result,
)
)
self.spf_results_combined.append(
"{0} / {1} / {2}".format(scope, domain, result)
)
self.spf_results_combined.append(f"{scope} / {domain} / {result}")
def save(self, **kwargs): # pyright: ignore[reportIncompatibleMethodOverride]
self.passed_dmarc = False
@@ -593,14 +589,14 @@ def create_indexes(names: list[str], settings: dict[str, Any] | None = None):
# — it is never installed as a mapping. See issue #169 and the
# *_combined fields on _AggregateReportDoc.
if not index.exists():
logger.debug("Creating OpenSearch index: {0}".format(name))
logger.debug(f"Creating OpenSearch index: {name}")
if settings is None:
index.settings(number_of_shards=1, number_of_replicas=0)
else:
index.settings(**settings)
index.create()
except Exception as e:
raise OpenSearchError("OpenSearch error: {0}".format(e.__str__()))
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
def migrate_indexes(
@@ -662,7 +658,7 @@ def migrate_indexes(
fo_mapping = fo_mapping[doc][fo_field]["mapping"][fo]
fo_type = fo_mapping["type"]
if fo_type == "long":
new_index_name = "{0}-v{1}".format(aggregate_index_name, version)
new_index_name = f"{aggregate_index_name}-v{version}"
body = {
"properties": {
"published_policy.fo": {
@@ -820,11 +816,11 @@ def save_aggregate_report_to_opensearch(
end_date_query = Q(dict(range=dict(date_end=dict(lte=end_date))))
if index_suffix is not None:
search_index = "dmarc_aggregate_{0}*".format(index_suffix)
search_index = f"dmarc_aggregate_{index_suffix}*"
else:
search_index = "dmarc_aggregate*"
if index_prefix is not None:
search_index = "{0}{1}".format(index_prefix, search_index)
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query & domain_query
query = query & begin_date_query & end_date_query
@@ -836,18 +832,15 @@ def save_aggregate_report_to_opensearch(
existing = search.execute()
except Exception as error_:
raise OpenSearchError(
"OpenSearch's search for existing report \
error: {}".format(error_.__str__())
f"OpenSearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
raise AlreadySaved(
"An aggregate report ID {0} from {1} about {2} "
"with a date range of {3} UTC to {4} UTC already "
f"An aggregate report ID {report_id} from {org_name} about {domain} "
f"with a date range of {begin_date_human} UTC to {end_date_human} UTC already "
"exists in "
"OpenSearch".format(
report_id, org_name, domain, begin_date_human, end_date_human
)
"OpenSearch"
)
published_policy = _PublishedPolicy(
domain=aggregate_report["policy_published"]["domain"],
@@ -940,11 +933,11 @@ def save_aggregate_report_to_opensearch(
index = "dmarc_aggregate"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = f"{index_prefix}{index}"
index = "{0}-{1}".format(index, index_date)
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -954,7 +947,7 @@ def save_aggregate_report_to_opensearch(
try:
agg_doc.save()
except Exception as e:
raise OpenSearchError("OpenSearch error: {0}".format(e.__str__()))
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
def save_failure_report_to_opensearch(
@@ -1002,12 +995,12 @@ def save_failure_report_to_opensearch(
arrival_date_epoch_milliseconds = int(arrival_date.timestamp() * 1000)
if index_suffix is not None:
search_index = "dmarc_failure_{0}*,dmarc_forensic_{0}*".format(index_suffix)
search_index = f"dmarc_failure_{index_suffix}*,dmarc_forensic_{index_suffix}*"
else:
search_index = "dmarc_failure*,dmarc_forensic*"
if index_prefix is not None:
search_index = ",".join(
"{0}{1}".format(index_prefix, part) for part in search_index.split(",")
f"{index_prefix}{part}" for part in search_index.split(",")
)
search = Search(index=search_index)
q = Q(dict(match=dict(arrival_date=arrival_date_epoch_milliseconds)))
@@ -1059,8 +1052,8 @@ def save_failure_report_to_opensearch(
if len(existing) > 0:
raise AlreadySaved(
"A failure sample to {0} from {1} "
"with a subject of {2} and arrival date of {3} "
"A failure sample to {} from {} "
"with a subject of {} and arrival date of {} "
"already exists in "
"OpenSearch".format(to_, from_, subject, failure_report["arrival_date_utc"])
)
@@ -1119,14 +1112,14 @@ def save_failure_report_to_opensearch(
index = "dmarc_failure"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = f"{index_prefix}{index}"
if monthly_indexes:
index_date = arrival_date.strftime("%Y-%m")
else:
index_date = arrival_date.strftime("%Y-%m-%d")
index = "{0}-{1}".format(index, index_date)
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -1135,10 +1128,10 @@ def save_failure_report_to_opensearch(
try:
failure_doc.save()
except Exception as e:
raise OpenSearchError("OpenSearch error: {0}".format(e.__str__()))
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
except KeyError as e:
raise InvalidFailureReport(
"Failure report missing required field: {0}".format(e.__str__())
f"Failure report missing required field: {e.__str__()}"
)
@@ -1185,11 +1178,11 @@ def save_smtp_tls_report_to_opensearch(
end_date_query = Q(dict(match=dict(date_end=end_date)))
if index_suffix is not None:
search_index = "smtp_tls_{0}*".format(index_suffix)
search_index = f"smtp_tls_{index_suffix}*"
else:
search_index = "smtp_tls*"
if index_prefix is not None:
search_index = "{0}{1}".format(index_prefix, search_index)
search_index = f"{index_prefix}{search_index}"
search = Search(index=search_index)
query = org_name_query & report_id_query
query = query & begin_date_query & end_date_query
@@ -1199,8 +1192,7 @@ def save_smtp_tls_report_to_opensearch(
existing = search.execute()
except Exception as error_:
raise OpenSearchError(
"OpenSearch's search for existing report \
error: {}".format(error_.__str__())
f"OpenSearch's search for existing report error: {error_.__str__()}"
)
if len(existing) > 0:
@@ -1214,10 +1206,10 @@ def save_smtp_tls_report_to_opensearch(
index = "smtp_tls"
if index_suffix:
index = "{0}_{1}".format(index, index_suffix)
index = f"{index}_{index_suffix}"
if index_prefix:
index = "{0}{1}".format(index_prefix, index)
index = "{0}-{1}".format(index, index_date)
index = f"{index_prefix}{index}"
index = f"{index}-{index_date}"
index_settings = dict(
number_of_shards=number_of_shards, number_of_replicas=number_of_replicas
)
@@ -1246,7 +1238,7 @@ def save_smtp_tls_report_to_opensearch(
policy_domain_combined = policy.get("policy_domain") or "none"
policy_type_combined = policy.get("policy_type") or "none"
smtp_tls_doc.policies_combined.append(
"{0} / {1}".format(policy_domain_combined, policy_type_combined)
f"{policy_domain_combined} / {policy_type_combined}"
)
policy_doc = _SMTPTLSPolicyDoc(
policy_domain=policy["policy_domain"],
@@ -1299,7 +1291,7 @@ def save_smtp_tls_report_to_opensearch(
failure_reason_code=failure_reason_code,
)
smtp_tls_doc.failure_details_combined.append(
"{0} / {1} / {2} / {3} / {4} / {5}".format(
"{} / {} / {} / {} / {} / {}".format(
policy_domain_combined,
policy_type_combined,
failure_detail.get("result_type") or "none",
@@ -1316,7 +1308,7 @@ def save_smtp_tls_report_to_opensearch(
try:
smtp_tls_doc.save()
except Exception as e:
raise OpenSearchError("OpenSearch error: {0}".format(e.__str__()))
raise OpenSearchError(f"OpenSearch error: {e.__str__()}")
# Backward-compatible aliases
parsedmarc/postgres.py
+2 -4
View File
@@ -661,10 +661,8 @@ class PostgreSQLClient:
)
if cur.fetchone() is not None:
raise AlreadySaved(
"A failure report with subject {subj!r} arriving "
"at {date} has already been saved".format(
subj=sample_subject, date=arrival_date_utc
)
f"A failure report with subject {sample_subject!r} arriving "
f"at {arrival_date_utc} has already been saved"
)
cur.execute(
"""
parsedmarc/resources/maps/AGENTS.md
+1
View File
@@ -283,6 +283,7 @@ Workflow:
```python
from parsedmarc.utils import load_psl_overrides, get_base_domain
load_psl_overrides(offline=True)
assert get_base_domain("host01.netlify.app") == "netlify.app"
```
parsedmarc/s3.py
+1 -3
View File
@@ -92,9 +92,7 @@ class S3Client(object):
report_id,
)
logger.debug(
"Saving {0} report to s3://{1}/{2}".format(
report_type, self.bucket_name, object_path
)
f"Saving {report_type} report to s3://{self.bucket_name}/{object_path}"
)
object_metadata = {
k: v
parsedmarc/splunk.py
+12 -12
View File
@@ -46,8 +46,8 @@ class HECClient(object):
data before giving up
"""
parsed_url = urlparse(url)
self.url = "{0}://{1}/services/collector/event/1.0".format(
parsed_url.scheme, parsed_url.netloc
self.url = (
f"{parsed_url.scheme}://{parsed_url.netloc}/services/collector/event/1.0"
)
self.access_token = access_token.lstrip("Splunk ")
self.index = index
@@ -62,7 +62,7 @@ class HECClient(object):
self.session = httpx.Client(
headers={
"User-Agent": USER_AGENT,
"Authorization": "Splunk {0}".format(self.access_token),
"Authorization": f"Splunk {self.access_token}",
},
verify=self.verify,
follow_redirects=True,
@@ -88,7 +88,7 @@ class HECClient(object):
return
data = self._common_data.copy()
json_str = ""
json_lines: list[str] = []
for report in aggregate_reports:
for record in report["records"]:
new_report: dict[str, str | int | float | dict] = dict()
@@ -127,13 +127,13 @@ class HECClient(object):
)
data["time"] = timestamp
data["event"] = new_report.copy()
json_str += "{0}\n".format(json.dumps(data))
json_lines.append(f"{json.dumps(data)}\n")
if not self.verify:
logger.debug("Skipping certificate verification for Splunk HEC")
try:
response = self.session.post(
self.url, content=json_str, timeout=self.timeout
self.url, content="".join(json_lines), timeout=self.timeout
)
response = response.json()
except Exception as e:
@@ -159,7 +159,7 @@ class HECClient(object):
if len(failure_reports) < 1:
return
json_str = ""
json_lines: list[str] = []
for report in failure_reports:
data = self._common_data.copy()
data["sourcetype"] = "dmarc:failure"
@@ -170,13 +170,13 @@ class HECClient(object):
)
data["time"] = timestamp
data["event"] = report.copy()
json_str += "{0}\n".format(json.dumps(data))
json_lines.append(f"{json.dumps(data)}\n")
if not self.verify:
logger.debug("Skipping certificate verification for Splunk HEC")
try:
response = self.session.post(
self.url, content=json_str, timeout=self.timeout
self.url, content="".join(json_lines), timeout=self.timeout
)
response = response.json()
except Exception as e:
@@ -203,19 +203,19 @@ class HECClient(object):
return
data = self._common_data.copy()
json_str = ""
json_lines: list[str] = []
for report in reports:
data["sourcetype"] = "smtp:tls"
timestamp = human_timestamp_to_unix_timestamp(report["begin_date"])
data["time"] = timestamp
data["event"] = report.copy()
json_str += "{0}\n".format(json.dumps(data))
json_lines.append(f"{json.dumps(data)}\n")
if not self.verify:
logger.debug("Skipping certificate verification for Splunk HEC")
try:
response = self.session.post(
self.url, content=json_str, timeout=self.timeout
self.url, content="".join(json_lines), timeout=self.timeout
)
response = response.json()
except Exception as e:
parsedmarc/utils.py
+3 -3
View File
@@ -235,7 +235,7 @@ def query_dns(
"""
domain = str(domain).lower()
record_type = record_type.upper()
cache_key = "{0}_{1}".format(domain, record_type)
cache_key = f"{domain}_{record_type}"
if cache:
cached_records = cache.get(cache_key, None)
if isinstance(cached_records, list):
@@ -1077,7 +1077,7 @@ def is_mbox(path: str) -> bool:
if len(mbox.keys()) > 0:
_is_mbox = True
except Exception as e:
logger.debug("Error checking for MBOX file: {0}".format(e.__str__()))
logger.debug(f"Error checking for MBOX file: {e.__str__()}")
return _is_mbox
@@ -1229,7 +1229,7 @@ def parse_email(data: bytes | str, *, strip_attachment_payloads: bool = False) -
payload = str.encode(payload)
attachment["sha256"] = hashlib.sha256(payload).hexdigest()
except Exception as e:
logger.debug("Unable to decode attachment: {0}".format(e.__str__()))
logger.debug(f"Unable to decode attachment: {e.__str__()}")
if strip_attachment_payloads:
for attachment in parsed_email["attachments"]:
if "payload" in attachment:
parsedmarc/webhook.py
+1 -1
View File
@@ -63,7 +63,7 @@ class WebhookClient(object):
else:
self.session.post(webhook_url, content=payload, timeout=self.timeout)
except Exception as error_:
logger.error("Webhook Error: {0}".format(error_.__str__()))
logger.error(f"Webhook Error: {error_.__str__()}")
def close(self):
"""Close the underlying HTTP session."""
pyproject.toml
+24 -8
View File
@@ -83,7 +83,7 @@ build = [
# Pinned exactly: pyright's checks evolve between releases, so an
# unpinned version could break CI without any code change. Bump
# deliberately (and fix any new findings) rather than implicitly.
"pyright==1.1.410",
"pyright==1.1.411",
"pytest",
"pytest-cov",
# Used only by the out-of-wheel maintainer script
@@ -92,11 +92,11 @@ build = [
# urllib3's HTTPAdapter machinery.
"requests>=2.22.0",
# Pinned exactly for the same reason as pyright: ruff's default rule
# set evolves between releases (0.16.0 began flagging this codebase's
# str.format() style), so an unpinned version breaks CI without any
# code change. Bump deliberately and fix any new findings in the same
# PR as the bump.
"ruff==0.15.21",
# set evolves between releases (e.g. 0.16.0 began flagging the
# str.format() style this codebase used until then), so an unpinned
# version breaks CI without any code change. Bump deliberately and fix
# any new findings in the same PR as the bump.
"ruff==0.16.0",
"sphinx",
"sphinx_rtd_theme",
]
@@ -130,13 +130,29 @@ exclude = [
]
[tool.ruff.lint]
# Enforce modern type-hint syntax on top of ruff's default rules. With
# The rule set is selected explicitly rather than floating on ruff's
# defaults: ruff 0.16.0 expanded the default selection from the
# long-standing E4/E7/E9/F to many more rule families (BLE, SIM, C4, DTZ,
# I, PL, S, ...), some of which conflict with deliberate house style —
# e.g. BLE001 flags the parser's intentional broad catches that keep one
# malformed report from crashing a batch. E4/E7/E9/F is the pre-0.16
# default set; adopting any of the new families is a deliberate
# per-family decision, made here with a comment, not an upgrade side
# effect.
#
# UP006/UP007/UP035/UP045 enforce modern type-hint syntax: with
# requires-python >=3.10, PEP 585 builtins (list[int]) and PEP 604 unions
# (X | Y, X | None) are available, so keep the deprecated typing.List /
# Union / Optional spellings out of the codebase.
extend-select = [
select = [
"E4", # import rules from pycodestyle
"E7", # statement rules from pycodestyle
"E9", # runtime/syntax error rules from pycodestyle
"F", # pyflakes
"UP006", # non-pep585-annotation: List -> list, Dict -> dict
"UP007", # non-pep604-annotation-union: Union[X, Y] -> X | Y
"UP030", # format-literals: "{0}".format(x) -> "{}".format(x)
"UP032", # f-string: "{}".format(x) -> f"{x}"
"UP035", # deprecated-import: typing.List etc. / typing -> collections.abc
"UP045", # non-pep604-annotation-optional: Optional[X] -> X | None
]
tests/test_init.py
+13 -13
View File
@@ -73,7 +73,7 @@ class Test(unittest.TestCase):
file = "samples/extract_report/nice-input.xml"
with open(file, "rb") as f:
data = f.read()
print("Testing {0}: ".format(file), end="")
print(f"Testing {file}: ", end="")
xmlout = parsedmarc.extract_report(data)
with open("samples/extract_report/nice-input.xml") as f:
xmlin = f.read()
@@ -84,7 +84,7 @@ class Test(unittest.TestCase):
"""Test extract report function for XML input"""
print()
file = "samples/extract_report/nice-input.xml"
print("Testing {0}: ".format(file), end="")
print(f"Testing {file}: ", end="")
xmlout = parsedmarc.extract_report_from_file_path(file)
with open("samples/extract_report/nice-input.xml") as f:
xmlin = f.read()
@@ -103,7 +103,7 @@ class Test(unittest.TestCase):
"""Test extract report function for gzip input"""
print()
file = "samples/extract_report/nice-input.xml.gz"
print("Testing {0}: ".format(file), end="")
print(f"Testing {file}: ", end="")
xmlout = parsedmarc.extract_report_from_file_path(file)
with open("samples/extract_report/nice-input.xml") as f:
xmlin = f.read()
@@ -114,7 +114,7 @@ class Test(unittest.TestCase):
"""Test extract report function for zip input"""
print()
file = "samples/extract_report/nice-input.xml.zip"
print("Testing {0}: ".format(file), end="")
print(f"Testing {file}: ", end="")
xmlout = parsedmarc.extract_report_from_file_path(file)
with open("samples/extract_report/nice-input.xml") as f:
xmlin = minify_xml(f.read())
@@ -155,7 +155,7 @@ class Test(unittest.TestCase):
for sample_path in sample_paths:
if os.path.isdir(sample_path):
continue
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
with self.subTest(sample=sample_path):
result = parsedmarc.parse_report_file(
sample_path, always_use_local_files=True, offline=OFFLINE_MODE
@@ -195,7 +195,7 @@ class Test(unittest.TestCase):
print()
sample_paths = glob("samples/failure/*.eml")
for sample_path in sample_paths:
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
with self.subTest(sample=sample_path):
with open(sample_path) as sample_file:
sample_content = sample_file.read()
@@ -245,7 +245,7 @@ class Test(unittest.TestCase):
"""Test parsing the sample report from RFC 9990 Appendix B"""
print()
sample_path = "samples/aggregate/rfc9990-sample.xml"
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
result = parsedmarc.parse_report_file(
sample_path, always_use_local_files=True, offline=True
)
@@ -323,7 +323,7 @@ class Test(unittest.TestCase):
sample_path = (
"samples/aggregate/example.net!example.com!1529366400!1529452799.xml"
)
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
result = parsedmarc.parse_report_file(
sample_path, always_use_local_files=True, offline=True
)
@@ -350,7 +350,7 @@ class Test(unittest.TestCase):
"samples/aggregate/"
"rfc9990-example.net!example.com!1700000000!1700086399.xml"
)
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
result = parsedmarc.parse_report_file(
sample_path, always_use_local_files=True, offline=True
)
@@ -590,7 +590,7 @@ class Test(unittest.TestCase):
for sample_path in sample_paths:
if os.path.isdir(sample_path):
continue
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
with self.subTest(sample=sample_path):
result = parsedmarc.parse_report_file(sample_path, offline=OFFLINE_MODE)
assert result["report_type"] == "smtp_tls"
@@ -1658,7 +1658,7 @@ class Test(unittest.TestCase):
"""parse_aggregate_report_file parses bytes input directly"""
print()
sample_path = "samples/aggregate/rfc9990-sample.xml"
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
with open(sample_path, "rb") as f:
data = f.read()
report = parsedmarc.parse_aggregate_report_file(
@@ -1677,7 +1677,7 @@ class Test(unittest.TestCase):
for sample_path in sample_paths:
if os.path.isdir(sample_path):
continue
print("Testing {0}: ".format(sample_path), end="")
print(f"Testing {sample_path}: ", end="")
with self.subTest(sample=sample_path):
parsed_report = cast(
AggregateReport,
@@ -1702,7 +1702,7 @@ class Test(unittest.TestCase):
print()
sample_paths = glob("samples/failure/*.eml")
for sample_path in sample_paths:
print("Testing CSV for {0}: ".format(sample_path), end="")
print(f"Testing CSV for {sample_path}: ", end="")
with self.subTest(sample=sample_path):
parsed_report = cast(
FailureReport,