🛡️ Sentinel: [CRITICAL/HIGH] Fix OAuth2 delimiter injection in IMAP/SMTP auth

.jules/sentinel.md

@@ -129,3 +129,7 @@
 **Vulnerability:** The URL validation logic correctly blocked non-global IP addresses and `localhost`, but failed to block internal domain extensions such as `.internal` or `.local` (or exact matches for `internal`). This could allow attackers to bypass SSRF protections by resolving these internal top-level domains.
 **Learning:** Checking for `localhost` alone is insufficient to prevent SSRF against internal network resources, as modern environments and protocols utilize `.internal` and `.local` domains for internal routing.
 **Prevention:** Always explicitly check and block domains matching `.internal`, `.local`, or `internal` (alongside `localhost`) when validating URLs for global reachability to prevent SSRF bypasses.
+## 2026-08-09 - OAuth2 Delimiter Injection Fix
+**Vulnerability:** OAuth2 string generation for IMAP/SMTP was vulnerable to delimiter injection by passing the `\x01` character in the user or token parameters.
+**Learning:** Attacker-controlled input inserted into protocols with specific delimiter characters can lead to parameter pollution or auth bypass.
+**Prevention:** Always validate or sanitize untrusted inputs specifically looking for protocol control characters like `\x01` in XOAUTH2 before encoding.

backend/services/email_client.py

@@ -45,6 +45,7 @@ class ValidatedSmtpDestination:
     proto: int
     sockaddr: tuple[Any, ...]
 
+
 @dataclass(frozen=True)
 class EmailMessageParams:
     to_address: str
@@ -53,6 +54,7 @@ class EmailMessageParams:
     in_reply_to: str | None = None
     references: str | None = None
 
+
 @dataclass(frozen=True)
 class SmtpConfig:
     smtp_server: str
@@ -61,9 +63,10 @@ class SmtpConfig:
     smtp_password: str | None = None
 
 
-
 def generate_oauth2_string(user: str, access_token: str) -> bytes:
     """Generates an OAuth2 string for IMAP/SMTP authentication."""
+    if "\x01" in user or "\x01" in access_token:
+        raise ValueError("Invalid character in user or access token")
     auth_string = f"user={user}\x01auth=Bearer {access_token}\x01\x01"
     return base64.b64encode(auth_string.encode("utf-8"))
 
@@ -102,21 +105,15 @@ def _parse_allowed_ports(
 
 
 def _parse_allowed_smtp_ports() -> set[int]:
-    return _parse_allowed_ports(
-        settings.ALLOWED_SMTP_PORTS, "SMTP", SMTP_EGRESS_PORTS
-    )
+    return _parse_allowed_ports(settings.ALLOWED_SMTP_PORTS, "SMTP", SMTP_EGRESS_PORTS)
 
 
 def _parse_allowed_imap_ports() -> set[int]:
-    return _parse_allowed_ports(
-        settings.ALLOWED_IMAP_PORTS, "IMAP", IMAP_EGRESS_PORTS
-    )
+    return _parse_allowed_ports(settings.ALLOWED_IMAP_PORTS, "IMAP", IMAP_EGRESS_PORTS)
 
 
 def _parse_allowed_pop3_ports() -> set[int]:
-    return _parse_allowed_ports(
-        settings.ALLOWED_POP3_PORTS, "POP3", POP3_EGRESS_PORTS
-    )
+    return _parse_allowed_ports(settings.ALLOWED_POP3_PORTS, "POP3", POP3_EGRESS_PORTS)
 
 
 def _parse_allowed_smtp_hosts() -> set[str]:
@@ -561,7 +558,9 @@ def build_email_message(
     message["To"] = _validate_email_header_value(message_params.to_address)
     message["Subject"] = _validate_email_header_value(message_params.subject)
     if message_params.in_reply_to:
-        message["In-Reply-To"] = _validate_email_header_value(message_params.in_reply_to)
+        message["In-Reply-To"] = _validate_email_header_value(
+            message_params.in_reply_to
+        )
     if message_params.references:
         message["References"] = _validate_email_header_value(message_params.references)
     message.set_content(message_params.body)
@@ -591,7 +590,9 @@ async def send_email(
         )
         return {"status": "simulated", "simulated": True}
 
-    smtp_destination = validate_smtp_destination(smtp_config.smtp_server, smtp_config.smtp_port)
+    smtp_destination = validate_smtp_destination(
+        smtp_config.smtp_server, smtp_config.smtp_port
+    )
 
     try:
         smtp_socket = await _connect_validated_smtp_socket(smtp_destination)

backend/tests/test_email_client.py

@@ -18,6 +18,14 @@ def test_generate_oauth2_string():
     assert b"auth=Bearer dummy_token" in decoded
 
 
+def test_generate_oauth2_string_rejects_delimiter():
+    with pytest.raises(ValueError, match="Invalid character in user or access token"):
+        generate_oauth2_string("test@example.com\x01test=test", "dummy_token")
+
+    with pytest.raises(ValueError, match="Invalid character in user or access token"):
+        generate_oauth2_string("test@example.com", "dummy_token\x01\x01")
+
+
 def test_build_email_message_sets_reply_headers():
     params = EmailMessageParams(
         to_address="test@example.com",
@@ -66,7 +74,9 @@ def test_build_email_message_rejects_newlines_in_header_fields(
         kwargs[field_name] = field_value
 
     params = EmailMessageParams(**kwargs)
-    with pytest.raises(ValueError, match="Email header fields must not contain newlines"):
+    with pytest.raises(
+        ValueError, match="Email header fields must not contain newlines"
+    ):
         build_email_message(message_params=params, from_address=from_address)
 
 
@@ -83,7 +93,10 @@ async def test_send_email_logs_sanitized_recipient(caplog):
 
     assert result == {"status": "simulated", "simulated": True}
     messages = [record.getMessage() for record in caplog.records]
-    assert "Simulating sending email to victim@example.com (no SMTP server configured)" in messages
+    assert (
+        "Simulating sending email to victim@example.com (no SMTP server configured)"
+        in messages
+    )
     assert all("\n" not in message and "\r" not in message for message in messages)