From d4cd6d60f14d2ef7dbf4f45ab6b51f7f16dbe797 Mon Sep 17 00:00:00 2001 From: Victor <124416658+victorv2i@users.noreply.github.com> Date: Fri, 26 Jun 2026 14:32:56 -0400 Subject: [PATCH] fix(email): validate IMAP/SMTP ports instead of crashing with 500 (#4464) The email-account endpoints coerced user-supplied ports with a bare int(data.get("imap_port") or 993), so a non-numeric port (e.g. "imap") raised ValueError and surfaced as an HTTP 500 in the create, update, and test-config endpoints. Add a _coerce_port(value, default) -> (port, error) helper and use it in all three endpoints, returning the endpoints standard {"ok": False, "error": ...} response (matching the existing "name required" validation) instead of crashing. A blank or missing port still falls back to the default (993/465). --- routes/email_routes.py | 44 +++++++++++++--- tests/test_email_account_port_validation.py | 56 +++++++++++++++++++++ 2 files changed, 92 insertions(+), 8 deletions(-) create mode 100644 tests/test_email_account_port_validation.py diff --git a/routes/email_routes.py b/routes/email_routes.py index 77be0cdeb..2d3cc86e3 100644 --- a/routes/email_routes.py +++ b/routes/email_routes.py @@ -64,6 +64,21 @@ ODYSSEUS_MAIL_ORIGIN = "odysseus-ui" EMAIL_READ_ATTACHMENT_VERSION = 2 +def _coerce_port(value, default): + """Coerce a user-supplied port to int. + + Returns ``(port, error)``. A missing or blank value yields ``default``; a + non-numeric value yields ``(None, message)`` so callers can return a clean + error instead of letting ``int()`` raise and surface as an HTTP 500. + """ + if value in (None, ""): + return default, None + try: + return int(value), None + except (TypeError, ValueError): + return None, f"Invalid port {value!r}; must be a whole number" + + def _email_tag_owner_aliases(account_id: str | None, owner: str = "") -> list[str]: aliases = [owner or ""] try: @@ -3329,6 +3344,12 @@ def setup_email_routes(): name = (data.get("name") or "").strip() if not name: return {"ok": False, "error": "name required"} + imap_port, port_err = _coerce_port(data.get("imap_port"), 993) + if port_err: + return {"ok": False, "error": port_err} + smtp_port, port_err = _coerce_port(data.get("smtp_port"), 465) + if port_err: + return {"ok": False, "error": port_err} db = SessionLocal() try: row = EmailAccount( @@ -3337,13 +3358,13 @@ def setup_email_routes(): is_default=bool(data.get("is_default", False)), enabled=bool(data.get("enabled", True)), imap_host=(data.get("imap_host") or "").strip(), - imap_port=int(data.get("imap_port") or 993), + imap_port=imap_port, imap_user=(data.get("imap_user") or "").strip(), imap_password=_enc(data.get("imap_password") or ""), imap_starttls=bool(data.get("imap_starttls", True)), smtp_host=(data.get("smtp_host") or "").strip(), - smtp_port=int(data.get("smtp_port") or 465), - smtp_security=_smtp_security_mode({"smtp_security": data.get("smtp_security"), "smtp_port": data.get("smtp_port") or 465}), + smtp_port=smtp_port, + smtp_security=_smtp_security_mode({"smtp_security": data.get("smtp_security"), "smtp_port": smtp_port}), smtp_user=(data.get("smtp_user") or "").strip(), smtp_password=_enc(data.get("smtp_password") or ""), from_address=(data.get("from_address") or "").strip(), @@ -3387,7 +3408,10 @@ def setup_email_routes(): setattr(row, key, (data[key] or "").strip()) for key in ("imap_port", "smtp_port"): if data.get(key) not in (None, ""): - setattr(row, key, int(data[key])) + port, port_err = _coerce_port(data.get(key), None) + if port_err: + return {"ok": False, "error": port_err} + setattr(row, key, port) if "smtp_security" in data: row.smtp_security = _smtp_security_mode({"smtp_security": data.get("smtp_security"), "smtp_port": data.get("smtp_port") or row.smtp_port}) for key in ("imap_starttls", "enabled"): @@ -3491,12 +3515,14 @@ def setup_email_routes(): smtp_result = None imap_host = (body.get("imap_host") or "").strip() - imap_port = int(body.get("imap_port") or 993) + imap_port, imap_port_err = _coerce_port(body.get("imap_port"), 993) imap_user = (body.get("imap_user") or "").strip() imap_pass = body.get("imap_password") or "" imap_starttls = bool(body.get("imap_starttls")) - if not (imap_host and imap_user and imap_pass): + if imap_port_err: + imap_result = {"ok": False, "error": imap_port_err} + elif not (imap_host and imap_user and imap_pass): imap_result = {"ok": False, "error": "Need IMAP host, username, and password"} else: # Connection mode resolution: @@ -3523,8 +3549,10 @@ def setup_email_routes(): imap_result = {"ok": False, "error": _friendly_email_auth_error("IMAP", imap_host, e)} smtp_host = (body.get("smtp_host") or "").strip() - if smtp_host: - smtp_port = int(body.get("smtp_port") or 465) + smtp_port, smtp_port_err = _coerce_port(body.get("smtp_port"), 465) + if smtp_host and smtp_port_err: + smtp_result = {"ok": False, "error": smtp_port_err} + elif smtp_host: smtp_security = _smtp_security_mode({"smtp_security": body.get("smtp_security"), "smtp_port": smtp_port}) smtp_user = (body.get("smtp_user") or imap_user).strip() smtp_pass = body.get("smtp_password") or imap_pass diff --git a/tests/test_email_account_port_validation.py b/tests/test_email_account_port_validation.py new file mode 100644 index 000000000..d668e036b --- /dev/null +++ b/tests/test_email_account_port_validation.py @@ -0,0 +1,56 @@ +"""User-supplied IMAP/SMTP ports must not crash the email-account endpoints. + +A non-numeric port (for example ``"imap"`` or ``"993x"``) previously reached an +unguarded ``int(...)`` in create / update / test-config and raised ``ValueError``, +which surfaces as an HTTP 500. The endpoints should reject it with their standard +``{"ok": False, "error": ...}`` response instead. +""" + +import pytest + + +def _route_endpoint(router, path: str, method: str): + method = method.upper() + for route in router.routes: + if route.path == path and method in getattr(route, "methods", set()): + return route.endpoint + raise AssertionError(f"route not found: {method} {path}") + + +def test_coerce_port_accepts_int_and_numeric_string(): + import routes.email_routes as email_routes + assert email_routes._coerce_port(2525, 993) == (2525, None) + assert email_routes._coerce_port("465", 993) == (465, None) + + +def test_coerce_port_blank_uses_default(): + import routes.email_routes as email_routes + assert email_routes._coerce_port(None, 993) == (993, None) + assert email_routes._coerce_port("", 465) == (465, None) + + +def test_coerce_port_rejects_non_numeric(): + import routes.email_routes as email_routes + port, err = email_routes._coerce_port("imap", 993) + assert port is None + assert err and "port" in err.lower() + + +@pytest.mark.asyncio +async def test_create_account_rejects_non_numeric_port(): + """A bad port is rejected before any DB work, with the endpoint's error shape.""" + import routes.email_routes as email_routes + router = email_routes.setup_email_routes() + create = _route_endpoint(router, "/api/email/accounts", "POST") + result = await create( + { + "name": "Test", + "imap_host": "mail.example.com", + "imap_user": "u", + "imap_password": "p", + "imap_port": "not-a-number", + }, + owner="alice", + ) + assert result["ok"] is False + assert "port" in result["error"].lower()