Subchapter 42.4

Read Mailcatcher

Perform Preflight

Running Work Transitions

Threat Modeling

Triaging Security Findings

def attempt

def main

This file

Number: 42.4
Position: 4 of 8
Type: Python
Size: 13 KB
Lines: 337

Nearby

Source

scripts/read_mailcatcher.py

Python337 lines13 KB

1#!/usr/bin/env python3
2"""Fetch a Bitwarden Mailcatcher message matching a recipient + subject, and
3print the first matching URL from its body on stdout.
4
5Exit 0 on success (prints the extracted URL on stdout).
6Exit 1: no matching message found, or a matched message had no local-host URL
7        (diagnostic on stderr).
8Exit 2: usage error.
9Exit 3: Mailcatcher unreachable or returned invalid JSON.
10
11Usage:
12  read_mailcatcher.py --recipient <email> [--pattern <subject-keyword>]
13                      [--link-filter <regex>]
14
15--pattern is optional. Omit (or pass empty) to match any subject and just take

read_mailcatcher.py on GitHub

the most recent message for the recipient.

The Mailcatcher endpoint is the same across Bitwarden dev environments, so the

base URL is fixed at http://localhost:1080 and cannot be overridden. Extracted

URLs are still restricted to a local dev allowlist; extend it (never replace it)

for a custom local hostname with the comma-separated MAILCATCHER_ALLOWED_HOSTS

env var.

Defaults:

Designed to be called via the Bash tool. ../references/manual-api-walkthrough.md

documents the underlying Mailcatcher REST API this wraps.

"""

import argparse

import html

import json

import os

import re

import sys

import time

import urllib.error

import urllib.parse

import urllib.request

DEFAULT_MAILCATCHER_URL = "http://localhost:1080"

# Local dev hosts allowed for extracted URLs. Extend (never replace) via the

# comma-separated env var MAILCATCHER_ALLOWED_HOSTS.

DEFAULT_ALLOWED_HOSTS = ("localhost", "127.0.0.1", "::1", "bitwarden.test")

# Whitespace terminates a URL. grep is line-based, so the bash original could

# never match across a newline; excluding all whitespace preserves that.

# Quotes and angle brackets end a URL inside an HTML attribute or tag, and a

# backslash ends one so a browser and urlparse never see different hosts.

URL_PATTERN = re.compile(r'https?://[^\s>"\'<)\\]+')

REQUEST_TIMEOUT = 5

RETRY_DELAY = 3

EXIT_OK = 0

EXIT_NO_MATCH = 1

EXIT_USAGE = 2

EXIT_UNREACHABLE = 3

class Unreachable(Exception):

"""Mailcatcher could not be reached or returned something unparseable."""

class _NoRedirectHandler(urllib.request.HTTPRedirectHandler):

"""Never follow 3xx redirects, matching curl's default (no -L).

The stdlib's default opener installs a HTTPRedirectHandler that follows a

301/302/303/307 Location header with no second check, which would let a

redirect return content from a host that was never validated. Returning

None from redirect_request tells the base handler not to redirect; it

then falls through to HTTPDefaultErrorHandler, which raises HTTPError for

the halted 3xx status. _http_get catches that HTTPError and returns the

3xx response's own body without ever requesting the Location target,

matching curl -fsS with no -L: -f only fails on status >= 400, so a

halted redirect is a success that yields the redirect response's body.

"""

def redirect_request(self, req, fp, code, msg, headers, newurl):

return None

def allowed_hosts(env):

"""Local dev hosts allowed for extracted URLs.

The defaults are extended (never replaced) by the comma-separated

MAILCATCHER_ALLOWED_HOSTS env var. That var is operator-supplied

(a real environment assignment, not model-constructed argv), so it stays

outside the auto-approved `script:*` Bash grant and is a safe way to widen

the allowlist for a custom local hostname.

"""

hosts = [host.lower() for host in DEFAULT_ALLOWED_HOSTS]

for entry in env.get("MAILCATCHER_ALLOWED_HOSTS", "").split(","):

entry = entry.strip().lower()

if entry:

hosts.append(entry)

return tuple(hosts)

def _http_get(url):

"""GET url and return the body as text. Raises Unreachable on failure.

Mirrors curl -fsS with no -L: -f fails only on an HTTP status >= 400.

A halted 3xx (see _NoRedirectHandler) is not a failure under -f, so it

is treated the same as any other success and its own body is returned;

the Location target is never requested. Status >= 400 raises

Unreachable, as does any transport failure.

No TLS bypass here: the bash original used plain `curl -fsS` with no

`-k`, so an https base gets normal certificate verification. The base is

the fixed local http://localhost:1080 endpoint, not a caller-supplied

value, so there is no untrusted host to validate before the request.

"""

opener = urllib.request.build_opener(_NoRedirectHandler)

try:

with opener.open(url, timeout=REQUEST_TIMEOUT) as response:

return response.read().decode("utf-8", errors="replace")

except urllib.error.HTTPError as err:

if 300 <= err.code < 400:

return err.read().decode("utf-8", errors="replace")

raise Unreachable(str(err))

except (urllib.error.URLError, OSError) as err:

raise Unreachable(str(err))

def fetch_messages(base):

body = _http_get(f"{base}/messages")

try:

messages = json.loads(body)

except ValueError as err:

raise Unreachable(f"invalid JSON from {base}/messages: {err}")

if not isinstance(messages, list):

raise Unreachable(f"expected a JSON array from {base}/messages")

return messages

def select_message(messages, recipient, pattern):

"""Id of the newest message matching the recipient and optional pattern.

A message whose `id` is missing or not an integer is dropped rather than

indexed into. Mailcatcher always sends an integer id, but keying `max` on

`msg["id"]` turned a missing id into a KeyError, and a mix of int and str

ids into a TypeError, both of which escaped this script's documented exit

codes; an unusable entry is simply not a candidate.

"""

recipient = recipient.lower()

pattern = (pattern or "").lower()

matches = [

msg

for msg in messages

if isinstance(msg, dict)

and isinstance(msg.get("id"), int)

and not isinstance(msg.get("id"), bool)

and any(recipient in str(entry).lower() for entry in msg.get("recipients", []))

and (not pattern or pattern in str(msg.get("subject", "")).lower())

]

if not matches:

return None

return max(matches, key=lambda msg: msg["id"])["id"]

def fetch_body(base, message_id):

"""Return the message body, preferring plain text and falling back to HTML.

Distinguishes a genuinely empty plain part (Mailcatcher reachable, no plain

body) from an outage that strikes between the message-list fetch and the

body fetch. If BOTH the plain and HTML fetches fail to reach Mailcatcher,

raise Unreachable so the caller reports EXIT_UNREACHABLE rather than

mislabeling an outage as a link-filter miss.

"""

try:

plain = _http_get(f"{base}/messages/{message_id}.plain")

except Unreachable:

plain = None

if plain:

return plain

if plain is not None:

# Reached Mailcatcher, but the plain part was empty: fall back to HTML.

print(

f"WARNING: plain body empty for message {message_id}; "

"using HTML body for URL extraction",

file=sys.stderr,

)

# HTML href attributes carry entity-encoded ampersands (`&`) between

# query params. Unescaping here, on the HTML branch only, keeps the plain

# path byte-exact while making the extracted link usable rather than corrupt

# from the first `&` onward. Whitespace-terminated URL matching is

# unaffected: `&`/`%40` are not whitespace either way.

try:

return html.unescape(_http_get(f"{base}/messages/{message_id}.html"))

except Unreachable:

if plain is None:

# Neither part could be reached: an outage, not an empty body.

raise

return ""

def matching_urls(body, link_filter):

"""Every URL in body matching link_filter, in document order.

The caller (attempt) applies the local-host preference on top of this: an

earlier non-local link that happens to match the filter (a marketing or

help footer link, say) must not mask a later local action link, so the

caller walks the full ordered list rather than taking the first match here.

"""

try:

return [

candidate

for candidate in URL_PATTERN.findall(body or "")

if re.search(link_filter, candidate, re.IGNORECASE)

]

except re.error:

# Mirrors grep -iE with a malformed ERE: it errors out and yields no

# match rather than a stack trace, so this falls through to the

# caller's existing no-URL-found NO_MATCH branch.

return []

def _hostname(url):

"""Lowercased hostname of url, or '' if it cannot be parsed."""

try:

return (urllib.parse.urlparse(url).hostname or "").lower()

except ValueError:

return ""

def is_local(url, allowed):

"""Whether url points at an allowed local dev host.

Browsers end the authority at a backslash and urlparse does not, so

"https://evil.com\\@localhost/" parses as localhost here but navigates to

evil.com. Dev email links carry neither a backslash nor userinfo, so either

one rejects the URL before the host is checked.

"""

if "\\" in url:

return False

try:

if "@" in urllib.parse.urlparse(url).netloc:

return False

except ValueError:

return False

return _hostname(url) in allowed

def attempt(base, recipient, pattern, link_filter, allowed):

"""One pass. Returns (code, url_or_None).

code 0 with a URL on success; 1 when no message matched (retryable);

2 when a message matched but yielded no usable URL (not retryable);

3 when Mailcatcher is unreachable (not retryable).

"""

try:

messages = fetch_messages(base)

except Unreachable as err:

print(f"ERROR: Mailcatcher unreachable at {base}: {err}", file=sys.stderr)

return 3, None

message_id = select_message(messages, recipient, pattern)

if message_id is None:

return 1, None

try:

body = fetch_body(base, message_id)

except Unreachable as err:

print(f"ERROR: Mailcatcher unreachable at {base}: {err}", file=sys.stderr)

return 3, None

candidates = matching_urls(body, link_filter)

if not candidates:

print(

f"NO_MATCH: message {message_id} matched but contained no URL "

f"filtered by '{link_filter}'",

file=sys.stderr,

)

return 2, None

url = next((c for c in candidates if is_local(c, allowed)), None)

if url is None:

rejected = sorted({_hostname(c) for c in candidates} - {""})

print(

f"NO_MATCH: none of the {len(candidates)} filter-matched URL(s) is a "

f"local dev host; rejected host(s): {', '.join(rejected) or '(unparseable)'}",

file=sys.stderr,

)

return 2, None

return 0, url

def main(argv, env):

parser = argparse.ArgumentParser(

prog="read_mailcatcher.py",

description="Print the first matching URL from a Mailcatcher message.",

)

parser.add_argument("--recipient", required=True)

parser.add_argument("--pattern", default="")

parser.add_argument("--link-filter", default=DEFAULT_LINK_FILTER)

try:

args = parser.parse_args(argv)

except SystemExit as exc:

return exc.code if isinstance(exc.code, int) else EXIT_USAGE

if not args.recipient.strip():

print("ERROR: --recipient is required", file=sys.stderr)

return EXIT_USAGE

# The Mailcatcher endpoint is identical across Bitwarden dev environments,

# so the base URL is fixed here rather than taken from argv or the

# environment. Keeping it out of the auto-approved `script:*` argv shape

# means a granted call can never point the fetch at an arbitrary host.

base = DEFAULT_MAILCATCHER_URL

allowed = allowed_hosts(env)

settled = {3: EXIT_UNREACHABLE, 2: EXIT_NO_MATCH}

code, url = attempt(base, args.recipient, args.pattern, args.link_filter, allowed)

if code == 0:

print(url)

return EXIT_OK

if code in settled:

return settled[code]

# code was 1: Mailcatcher may not have received the message yet. Retry once.

time.sleep(RETRY_DELAY)

code, url = attempt(base, args.recipient, args.pattern, args.link_filter, allowed)

if code == 0:

print(url)

return EXIT_OK

if code in settled:

return settled[code]

if args.pattern:

print(

f"NO_MATCH: no email for recipient '{args.recipient}' "

f"with subject containing '{args.pattern}'",

file=sys.stderr,

)

else:

print(f"NO_MATCH: no email for recipient '{args.recipient}'", file=sys.stderr)

return EXIT_NO_MATCH

if __name__ == "__main__":

sys.exit(main(sys.argv[1:], os.environ))