aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
authorDanilo M. <danix@danix.xyz>2026-09-08 13:47:42 +0200
committerDanilo M. <danix@danix.xyz>2026-09-08 13:47:42 +0200
commitab8be8e3d58a5c2dd8b351110822063b482311e1 (patch)
treeb120813a6fc6d5b2f679c9122c75ce7ba1ebcd9c
parente7b0a533bb92fd0a800f92d220aacf994cef0387 (diff)
downloadabusectl-ab8be8e3d58a5c2dd8b351110822063b482311e1.tar.gz
abusectl-ab8be8e3d58a5c2dd8b351110822063b482311e1.zip
refactor: one URL scan serving both the report and the flagging
The IOC assembly needed each URL in two forms: redacted for the report, original to recognise a suspect token, since by the time a query value reads REDACTED there is nothing left to look at. That arrived as a second scanner rebuilding a {redacted: original} mapping by repeating urls()'s own walk, which is two functions that have to stay in step by hand. One private _scan_urls() returns both forms instead. urls() keeps its signature and is now a sort over its keys, and the two forms cannot drift because nothing derives one from the other twice. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01KphFXTc2QajxXsHWyvGJ4R
-rw-r--r--abusectl/parse.py53
1 files changed, 24 insertions, 29 deletions
diff --git a/abusectl/parse.py b/abusectl/parse.py
index eaf6165..79cccc9 100644
--- a/abusectl/parse.py
+++ b/abusectl/parse.py
@@ -226,21 +226,33 @@ def _text_parts(message) -> list[str]:
return parts
-def urls(raw: bytes) -> list[str]:
- """Return every http(s) URL found in the message's text parts, redacted.
+def _scan_urls(raw: bytes) -> dict[str, str]:
+ """Return {redacted: original} for every http(s) URL in the text parts.
+
+ ONE scan serving both callers. The redacted form is what gets reported;
+ the original is needed to recognise a suspect token, because by the time
+ a query value reads REDACTED there is nothing left to look at. Keeping
+ them together means the two forms cannot drift, which a second scanning
+ function would eventually let them do.
Nothing here is fetched or resolved: URLs are found by matching the
- message's own text, never by requesting anything. Each match is
- redacted with redact.url() before being returned, deduplicated and
- sorted so the result is stable regardless of where in the message a
- URL happened to repeat.
+ message's own text, never by requesting anything.
"""
- message = _message(raw)
- found = set()
- for text in _text_parts(message):
+ found = {}
+ for text in _text_parts(_message(raw)):
for match in _URL.findall(text):
- found.add(redact.url(match.rstrip(".,;:!?")))
- return sorted(found)
+ original = match.rstrip(".,;:!?")
+ found[redact.url(original)] = original
+ return found
+
+
+def urls(raw: bytes) -> list[str]:
+ """Return every http(s) URL found in the message's text parts, redacted.
+
+ Deduplicated and sorted, so the result is stable regardless of where in
+ the message a URL happened to repeat.
+ """
+ return sorted(_scan_urls(raw))
def _redirect_targets(url_text: str, depth: int, seen: set[str]) -> list[tuple[str, str]]:
@@ -321,23 +333,6 @@ def _suspect_segments(raw_url: str) -> list[str]:
return found
-def _original_urls(message) -> dict[str, str]:
- """Return {redacted_url: original_url} for every URL in the message.
-
- urls() returns only the redacted form, which is right for the report
- but useless for flagging a suspect query value: by the time a value is
- REDACTED there is nothing left to look at. This mirrors urls()'s own
- scan so the two stay in step, keyed by the redacted form since that is
- what the caller already has in hand.
- """
- mapping = {}
- for text in _text_parts(message):
- for match in _URL.findall(text):
- original = match.rstrip(".,;:!?")
- mapping[redact.url(original)] = original
- return mapping
-
-
def iocs(raw: bytes, trusted: list[str]) -> list[dict]:
"""Assemble every indicator this module can extract into one flat list.
@@ -365,7 +360,7 @@ def iocs(raw: bytes, trusted: list[str]) -> list[dict]:
NoTrustBoundary before any other extraction runs.
"""
boundary_ip = sending_ip(raw, trusted)
- originals = _original_urls(_message(raw))
+ originals = _scan_urls(raw)
result: list[dict] = []