diff options
| author | Danilo M. <danix@danix.xyz> | 2026-09-08 13:47:42 +0200 |
|---|---|---|
| committer | Danilo M. <danix@danix.xyz> | 2026-09-08 13:47:42 +0200 |
| commit | ab8be8e3d58a5c2dd8b351110822063b482311e1 (patch) | |
| tree | b120813a6fc6d5b2f679c9122c75ce7ba1ebcd9c | |
| parent | e7b0a533bb92fd0a800f92d220aacf994cef0387 (diff) | |
| download | abusectl-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.py | 53 |
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] = [] |
