nirpa
e4d6230dd1
Services that need to reach a person hand the message here instead of each carrying a SendGrid key. Email goes to SendGrid, SMS to Twilio; whatsapp and push answer 501 rather than pretending. It binds loopback and the docker bridge only. An authenticated relay on the public internet is an open spam relay the moment the token leaks, and that token is copied into every consumer's environment. The log names a channel, a redacted address and the provider's answer. Never the subject, never the body.
119 lines
4.3 KiB
Python
119 lines
4.3 KiB
Python
"""What actually goes on the wire to SendGrid and Twilio.
|
|
|
|
Asserted at the seam rather than mocked at the module boundary, because the
|
|
shape of these two requests is the part a provider silently rejects: a From
|
|
that is not the verified sender, a field Twilio spells `To` and not `to`.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import json
|
|
import urllib.parse
|
|
|
|
from app import providers
|
|
|
|
|
|
def recorder(status: int, body: str = "{}", reply_headers: dict | None = None):
|
|
calls: list[dict] = []
|
|
|
|
def fake_post(url: str, *, data: bytes, headers: dict, timeout: float):
|
|
calls.append({"url": url, "data": data, "headers": headers, "timeout": timeout})
|
|
return providers.Response(status, body, reply_headers or {})
|
|
|
|
fake_post.calls = calls # type: ignore[attr-defined]
|
|
return fake_post
|
|
|
|
|
|
def test_email_is_addressed_from_the_verified_sender() -> None:
|
|
post = recorder(202, "", {"X-Message-Id": "abc"})
|
|
outcome = providers.send_email(
|
|
to="[email protected]",
|
|
subject="Due",
|
|
body="Bins",
|
|
api_key="sg-key",
|
|
sender="[email protected]",
|
|
timeout=5,
|
|
post=post,
|
|
)
|
|
sent = json.loads(post.calls[0]["data"])
|
|
assert post.calls[0]["url"] == providers.SENDGRID_URL
|
|
assert sent["from"]["email"] == "[email protected]"
|
|
assert sent["personalizations"][0]["to"] == [{"email": "[email protected]"}]
|
|
assert sent["content"][0]["type"] == "text/plain"
|
|
assert post.calls[0]["headers"]["Authorization"] == "Bearer sg-key"
|
|
assert outcome.ok and outcome.status == 202 and outcome.message_id == "abc"
|
|
|
|
|
|
def test_an_email_with_no_subject_still_sends() -> None:
|
|
post = recorder(202)
|
|
providers.send_email(
|
|
to="[email protected]", subject="", body="Bins", api_key="k", sender="[email protected]", timeout=5, post=post
|
|
)
|
|
assert json.loads(post.calls[0]["data"])["subject"] == "Notification"
|
|
|
|
|
|
def test_sendgrid_refusal_is_reported_not_raised() -> None:
|
|
post = recorder(403, json.dumps({"errors": [{"message": "The from address does not match a verified Sender"}]}))
|
|
outcome = providers.send_email(
|
|
to="[email protected]", subject="s", body="b", api_key="k", sender="[email protected]", timeout=5, post=post
|
|
)
|
|
assert not outcome.ok
|
|
assert outcome.status == 403
|
|
assert "verified Sender" in outcome.detail
|
|
|
|
|
|
def test_sms_is_form_encoded_with_twilio_field_names() -> None:
|
|
post = recorder(201, json.dumps({"sid": "SM9"}))
|
|
outcome = providers.send_sms(
|
|
to="+15551234567",
|
|
subject="Due",
|
|
body="Bins",
|
|
account_sid="AC123",
|
|
auth_token="secret",
|
|
sender="+15550001111",
|
|
timeout=5,
|
|
post=post,
|
|
)
|
|
fields = dict(urllib.parse.parse_qsl(post.calls[0]["data"].decode()))
|
|
assert post.calls[0]["url"].endswith("/Accounts/AC123/Messages.json")
|
|
assert fields == {"To": "+15551234567", "From": "+15550001111", "Body": "Due\n\nBins"}
|
|
expected = "Basic " + base64.b64encode(b"AC123:secret").decode()
|
|
assert post.calls[0]["headers"]["Authorization"] == expected
|
|
assert outcome.ok and outcome.message_id == "SM9"
|
|
|
|
|
|
def test_a_runaway_body_is_truncated_rather_than_billed_by_the_segment() -> None:
|
|
post = recorder(201, json.dumps({"sid": "SM9"}))
|
|
providers.send_sms(
|
|
to="+15551234567",
|
|
subject="",
|
|
body="x" * 50_000,
|
|
account_sid="AC1",
|
|
auth_token="t",
|
|
sender="+1555",
|
|
timeout=5,
|
|
post=post,
|
|
)
|
|
fields = dict(urllib.parse.parse_qsl(post.calls[0]["data"].decode()))
|
|
assert len(fields["Body"]) == providers.SMS_LIMIT
|
|
|
|
|
|
def test_twilio_refusal_carries_its_code() -> None:
|
|
post = recorder(400, json.dumps({"message": "To and From cannot be the same", "code": 21266}))
|
|
outcome = providers.send_sms(
|
|
to="+1555", subject="s", body="b", account_sid="AC1", auth_token="t", sender="+1555", timeout=5, post=post
|
|
)
|
|
assert not outcome.ok
|
|
assert outcome.status == 400
|
|
assert "21266" in outcome.detail
|
|
|
|
|
|
def test_an_unparseable_error_body_is_truncated_not_dumped() -> None:
|
|
post = recorder(500, "<html>" + "x" * 5000)
|
|
outcome = providers.send_email(
|
|
to="[email protected]", subject="s", body="b", api_key="k", sender="[email protected]", timeout=5, post=post
|
|
)
|
|
assert not outcome.ok
|
|
assert len(outcome.detail) <= 200
|