Mario Fetka
378 lines
14 KiB
Python
Executable File
378 lines
14 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
# This program is free software, licensed under the terms of the GNU GPL.
|
|
# See the Bongo COPYING file for full details.
|
|
# Copyright (c) 2026 Bongo Project contributors
|
|
|
|
"""Verify Bongo IMAP capabilities, TLS transitions, and authentication."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import base64
|
|
import os
|
|
import re
|
|
import select
|
|
import socket
|
|
import ssl
|
|
|
|
|
|
HOST = os.environ.get("BONGO_TEST_HOST", "127.0.0.1")
|
|
IMAP_PORT = int(os.environ.get("BONGO_TEST_IMAP_PORT", "143"))
|
|
IMAPS_PORT = int(os.environ.get("BONGO_TEST_IMAPS_PORT", "993"))
|
|
TIMEOUT = float(os.environ.get("BONGO_TEST_TIMEOUT", "10"))
|
|
USERNAME = os.environ.get("BONGO_TEST_USER", "test1")
|
|
PASSWORD = os.environ.get("BONGO_TEST_PASSWORD", "")
|
|
TAGGED = re.compile(rb"^([A-Za-z0-9]+) (OK|NO|BAD)(?: .*)?\r\n$")
|
|
|
|
BASE_CAPABILITIES = {
|
|
"IMAP4REV2", "IMAP4REV1", "IMAP4", "NAMESPACE", "ID", "IDLE",
|
|
"ENABLE", "UTF8=ACCEPT", "UIDPLUS", "MOVE", "SPECIAL-USE",
|
|
"CREATE-SPECIAL-USE", "LIST-EXTENDED", "LIST-STATUS", "STATUS=SIZE",
|
|
"QUOTA", "QUOTA=RES-STORAGE", "LITERAL+", "APPENDLIMIT=999999",
|
|
"XSENDER",
|
|
}
|
|
|
|
|
|
class IMAPCheckError(RuntimeError):
|
|
"""Raised when the live IMAP peer violates the tested contract."""
|
|
|
|
|
|
class IMAPConnection:
|
|
def __init__(self, port: int, implicit_tls: bool = False) -> None:
|
|
raw = socket.create_connection((HOST, port), timeout=TIMEOUT)
|
|
raw.settimeout(TIMEOUT)
|
|
self.socket: socket.socket | ssl.SSLSocket = raw
|
|
self.reader = raw.makefile("rb")
|
|
self.counter = 0
|
|
self.tls_version = ""
|
|
if implicit_tls:
|
|
self.start_tls()
|
|
greeting = self.line()
|
|
if not greeting.startswith(b"* OK "):
|
|
raise IMAPCheckError(f"unexpected IMAP greeting: {greeting!r}")
|
|
|
|
def close(self) -> None:
|
|
try:
|
|
self.reader.close()
|
|
finally:
|
|
self.socket.close()
|
|
|
|
def line(self) -> bytes:
|
|
line = self.reader.readline(65538)
|
|
if not line:
|
|
raise IMAPCheckError("connection closed while waiting for IMAP response")
|
|
if len(line) > 65536:
|
|
raise IMAPCheckError("IMAP response line exceeded 65536 bytes")
|
|
if not line.endswith(b"\r\n"):
|
|
raise IMAPCheckError(f"unterminated IMAP response: {line!r}")
|
|
return line
|
|
|
|
def start_tls(self) -> None:
|
|
self.reader.close()
|
|
context = ssl.create_default_context()
|
|
context.check_hostname = False
|
|
context.verify_mode = ssl.CERT_NONE
|
|
context.minimum_version = ssl.TLSVersion.TLSv1_2
|
|
self.socket = context.wrap_socket(self.socket, server_hostname=HOST)
|
|
self.socket.settimeout(TIMEOUT)
|
|
self.reader = self.socket.makefile("rb")
|
|
self.tls_version = self.socket.version() or ""
|
|
if self.tls_version not in {"TLSv1.2", "TLSv1.3"}:
|
|
raise IMAPCheckError(
|
|
f"negotiated unexpected TLS version {self.tls_version!r}"
|
|
)
|
|
|
|
def command(
|
|
self, command: str, expected: tuple[str, ...] = ("OK",)
|
|
) -> list[bytes]:
|
|
self.counter += 1
|
|
tag = f"T{self.counter:04d}"
|
|
self.socket.sendall(f"{tag} {command}\r\n".encode("utf-8"))
|
|
lines: list[bytes] = []
|
|
while True:
|
|
line = self.line()
|
|
match = TAGGED.match(line)
|
|
if match is not None and match.group(1).decode("ascii") == tag:
|
|
status = match.group(2).decode("ascii")
|
|
if status not in expected:
|
|
raise IMAPCheckError(
|
|
f"{command!r} expected {expected}, received {line!r}"
|
|
)
|
|
return lines + [line]
|
|
lines.append(line)
|
|
|
|
|
|
def quote(value: str) -> str:
|
|
return '"' + value.replace("\\", "\\\\").replace('"', '\\"') + '"'
|
|
|
|
|
|
def capabilities(lines: list[bytes]) -> set[str]:
|
|
for line in lines:
|
|
if line.upper().startswith(b"* CAPABILITY "):
|
|
return {
|
|
item.upper()
|
|
for item in line.decode("ascii", "strict").strip().split()[2:]
|
|
}
|
|
raise IMAPCheckError(f"CAPABILITY response missing in {lines!r}")
|
|
|
|
|
|
def assert_capabilities(
|
|
advertised: set[str], *, cleartext: bool, authenticated: bool = False
|
|
) -> None:
|
|
missing = BASE_CAPABILITIES - advertised
|
|
if missing:
|
|
raise IMAPCheckError(f"missing IMAP capabilities: {sorted(missing)}")
|
|
if cleartext:
|
|
if "STARTTLS" not in advertised or "LOGINDISABLED" not in advertised:
|
|
raise IMAPCheckError(
|
|
"cleartext IMAP must advertise STARTTLS and LOGINDISABLED"
|
|
)
|
|
if any(item.startswith("AUTH=") for item in advertised):
|
|
raise IMAPCheckError("cleartext IMAP advertised authentication")
|
|
else:
|
|
if "STARTTLS" in advertised or "LOGINDISABLED" in advertised:
|
|
raise IMAPCheckError("TLS IMAP advertised a cleartext-only capability")
|
|
if not authenticated:
|
|
required = {"AUTH=PLAIN", "AUTH=LOGIN", "SASL-IR"}
|
|
if not required.issubset(advertised):
|
|
raise IMAPCheckError(
|
|
"missing TLS authentication capabilities: "
|
|
f"{sorted(required - advertised)}"
|
|
)
|
|
|
|
|
|
def check_starttls() -> str:
|
|
client = IMAPConnection(IMAP_PORT)
|
|
try:
|
|
assert_capabilities(capabilities(client.command("CAPABILITY")), cleartext=True)
|
|
client.command(
|
|
f"LOGIN {quote(USERNAME)} {quote(PASSWORD)}", expected=("NO",)
|
|
)
|
|
client.command("STARTTLS")
|
|
client.start_tls()
|
|
assert_capabilities(
|
|
capabilities(client.command("CAPABILITY")), cleartext=False
|
|
)
|
|
client.command("STARTTLS", expected=("NO", "BAD"))
|
|
client.command(f"LOGIN {quote(USERNAME)} {quote(PASSWORD)}")
|
|
client.command("LOGOUT")
|
|
return client.tls_version
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def check_starttls_pipeline_is_discarded() -> None:
|
|
client = IMAPConnection(IMAP_PORT)
|
|
try:
|
|
client.counter += 1
|
|
tag = f"T{client.counter:04d}"
|
|
client.socket.sendall(
|
|
f"{tag} STARTTLS\r\nT9999 LOGIN "
|
|
f"{quote(USERNAME)} {quote(PASSWORD)}\r\n".encode("utf-8")
|
|
)
|
|
line = client.line()
|
|
if not line.startswith(f"{tag} OK ".encode("ascii")):
|
|
raise IMAPCheckError(f"STARTTLS failed before pipeline test: {line!r}")
|
|
client.start_tls()
|
|
pending = (
|
|
client.socket.pending()
|
|
if isinstance(client.socket, ssl.SSLSocket)
|
|
else 0
|
|
)
|
|
readable, _, _ = select.select([client.socket], [], [], 0.5)
|
|
leaked = client.line() if pending or readable else b""
|
|
if leaked:
|
|
raise IMAPCheckError(
|
|
f"cleartext command crossed STARTTLS boundary: {leaked!r}"
|
|
)
|
|
client.command("LOGOUT")
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def auth_plain() -> str:
|
|
client = IMAPConnection(IMAPS_PORT, implicit_tls=True)
|
|
try:
|
|
assert_capabilities(
|
|
capabilities(client.command("CAPABILITY")), cleartext=False
|
|
)
|
|
payload = base64.b64encode(
|
|
b"\0" + USERNAME.encode() + b"\0" + PASSWORD.encode()
|
|
).decode("ascii")
|
|
client.command(f"AUTHENTICATE PLAIN {payload}")
|
|
client.command("LOGOUT")
|
|
return client.tls_version
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def auth_login() -> str:
|
|
client = IMAPConnection(IMAPS_PORT, implicit_tls=True)
|
|
try:
|
|
assert_capabilities(
|
|
capabilities(client.command("CAPABILITY")), cleartext=False
|
|
)
|
|
client.counter += 1
|
|
tag = f"T{client.counter:04d}"
|
|
client.socket.sendall(f"{tag} AUTHENTICATE LOGIN\r\n".encode("ascii"))
|
|
if not client.line().startswith(b"+ "):
|
|
raise IMAPCheckError("missing SASL LOGIN username challenge")
|
|
client.socket.sendall(base64.b64encode(USERNAME.encode()) + b"\r\n")
|
|
if not client.line().startswith(b"+ "):
|
|
raise IMAPCheckError("missing SASL LOGIN password challenge")
|
|
client.socket.sendall(base64.b64encode(PASSWORD.encode()) + b"\r\n")
|
|
completion = client.line()
|
|
if not completion.startswith(f"{tag} OK ".encode("ascii")):
|
|
raise IMAPCheckError(f"SASL LOGIN failed: {completion!r}")
|
|
client.command("LOGOUT")
|
|
return client.tls_version
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def auth_cancel() -> None:
|
|
client = IMAPConnection(IMAPS_PORT, implicit_tls=True)
|
|
try:
|
|
client.counter += 1
|
|
tag = f"T{client.counter:04d}"
|
|
client.socket.sendall(f"{tag} AUTHENTICATE LOGIN\r\n".encode("ascii"))
|
|
if not client.line().startswith(b"+ "):
|
|
raise IMAPCheckError("missing SASL challenge before cancellation")
|
|
client.socket.sendall(b"*\r\n")
|
|
completion = client.line()
|
|
if not completion.startswith(f"{tag} BAD ".encode("ascii")):
|
|
raise IMAPCheckError(
|
|
f"AUTHENTICATE cancellation was not tagged BAD: {completion!r}"
|
|
)
|
|
client.command("CAPABILITY")
|
|
client.command("LOGOUT")
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def auth_invalid_inputs() -> None:
|
|
client = IMAPConnection(IMAPS_PORT, implicit_tls=True)
|
|
try:
|
|
client.command("AUTHENTICATE DOES-NOT-EXIST", expected=("NO",))
|
|
client.command("AUTHENTICATE PLAIN not!base64", expected=("BAD",))
|
|
client.command("AUTHENTICATE PLAIN = extra", expected=("BAD",))
|
|
client.command("NOOP")
|
|
client.command("LOGOUT")
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def id_literal_plus(client: IMAPConnection) -> None:
|
|
name = b"Bongo literal client"
|
|
client.counter += 1
|
|
tag = f"T{client.counter:04d}"
|
|
client.socket.sendall(
|
|
f"{tag} ID ({{4+}}\r\n".encode("ascii")
|
|
+ b"name "
|
|
+ f"{{{len(name)}+}}\r\n".encode("ascii")
|
|
+ name
|
|
+ b")\r\n"
|
|
)
|
|
lines: list[bytes] = []
|
|
while True:
|
|
line = client.line()
|
|
if line.startswith(f"{tag} OK ".encode("ascii")):
|
|
break
|
|
if line.startswith(f"{tag} ".encode("ascii")):
|
|
raise IMAPCheckError(f"literal+ ID failed: {line!r}")
|
|
lines.append(line)
|
|
if not any(line.startswith(b'* ID ("name" "Bongo" ') for line in lines):
|
|
raise IMAPCheckError(f"literal+ ID response missing: {lines!r}")
|
|
|
|
|
|
def imaps_login() -> str:
|
|
client = IMAPConnection(IMAPS_PORT, implicit_tls=True)
|
|
try:
|
|
assert_capabilities(
|
|
capabilities(client.command("CAPABILITY")), cleartext=False
|
|
)
|
|
client.command(
|
|
f"LOGIN {quote(USERNAME)} {quote(PASSWORD)} extra",
|
|
expected=("BAD",),
|
|
)
|
|
client.command(f"LOGIN {quote(USERNAME)} {quote(PASSWORD)}")
|
|
assert_capabilities(
|
|
capabilities(client.command("CAPABILITY")),
|
|
cleartext=False,
|
|
authenticated=True,
|
|
)
|
|
client.command("CAPABILITY extra", expected=("BAD",))
|
|
client.command("NOOP")
|
|
client.command("NOOP extra", expected=("BAD",))
|
|
identity = client.command(
|
|
'ID ("name" "Bongo live test" "version" "0.7")'
|
|
)
|
|
if not any(
|
|
line.startswith(b'* ID ("name" "Bongo" "version" "')
|
|
for line in identity
|
|
):
|
|
raise IMAPCheckError(f"invalid ID response: {identity!r}")
|
|
id_literal_plus(client)
|
|
client.command('ID (NIL "value")', expected=("BAD",))
|
|
client.command('ID ("name""value")', expected=("BAD",))
|
|
client.command(
|
|
'ID ("name" "value""version" "0.7")', expected=("BAD",)
|
|
)
|
|
client.command(
|
|
'ID ("0123456789012345678901234567890" "value")',
|
|
expected=("BAD",),
|
|
)
|
|
client.command(
|
|
"ID ("
|
|
+ " ".join(f'"key{index}" "value"' for index in range(31))
|
|
+ ")",
|
|
expected=("BAD",),
|
|
)
|
|
client.command(
|
|
'ID ("comment" "' + ("x" * 1025) + '")',
|
|
expected=("BAD",),
|
|
)
|
|
client.command("NOOP")
|
|
namespace = client.command("NAMESPACE")
|
|
if b'* NAMESPACE (("" "/")) NIL NIL\r\n' not in namespace:
|
|
raise IMAPCheckError(f"invalid NAMESPACE response: {namespace!r}")
|
|
client.command("NAMESPACE extra", expected=("BAD",))
|
|
client.command("LOGOUT extra", expected=("BAD",))
|
|
client.command("NOOP")
|
|
logout = client.command("LOGOUT")
|
|
if not any(line.startswith(b"* BYE ") for line in logout):
|
|
raise IMAPCheckError(f"LOGOUT did not send BYE: {logout!r}")
|
|
return client.tls_version
|
|
finally:
|
|
client.close()
|
|
|
|
|
|
def main() -> int:
|
|
if not PASSWORD:
|
|
raise IMAPCheckError("BONGO_TEST_PASSWORD must be set")
|
|
starttls_version = check_starttls()
|
|
check_starttls_pipeline_is_discarded()
|
|
imaps_version = imaps_login()
|
|
if auth_plain() != imaps_version:
|
|
raise IMAPCheckError("IMAPS TLS version changed for SASL PLAIN")
|
|
if auth_login() != imaps_version:
|
|
raise IMAPCheckError("IMAPS TLS version changed for SASL LOGIN")
|
|
auth_cancel()
|
|
auth_invalid_inputs()
|
|
print(
|
|
"IMAP-01/02/03/04 PASS "
|
|
f"host={HOST} starttls={IMAP_PORT}/{starttls_version} "
|
|
f"imaps={IMAPS_PORT}/{imaps_version} "
|
|
"protocols=IMAP4rev2,IMAP4rev1,IMAP4 "
|
|
"pre-tls=LOGINDISABLED login=LOGIN,PLAIN,LOGIN-SASL "
|
|
"commands=CAPABILITY,NOOP,ID,NAMESPACE,LOGOUT "
|
|
"id=literal+,limits,nil-key-rejected "
|
|
"auth=cancel/unknown/base64/trailing/session-alive "
|
|
"noarg-syntax=capability/noop/namespace/logout "
|
|
"repeated-starttls=rejected pipeline=discarded"
|
|
)
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|