XF/GIA
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

Increase security and reformat

Browse Source
This commit is contained in:
Mark Veidemanis
2026-03-07 20:52:13 +00:00
parent 10588a18b9
commit bca4d6898f
144 changed files with 6735 additions and 3960 deletions
Download Patch File Download Diff File Expand all files Collapse all files

182
core/tests/test_xmpp_integration.py
View File

@@ -10,6 +10,7 @@ mirroring exactly the flow a phone XMPP client uses:
Tests are skipped automatically when XMPP settings are absent (e.g. in CI
environments without a running stack).
"""
from __future__ import annotations
import base64
@@ -26,11 +27,11 @@ import xml.etree.ElementTree as ET
from django.conf import settings
from django.test import SimpleTestCase
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _xmpp_configured() -> bool:
return bool(
getattr(settings, "XMPP_JID", None)
@@ -67,10 +68,21 @@ def _xmpp_domain() -> str:
def _prosody_auth_endpoint() -> str:
"""URL of the Django auth bridge that Prosody calls for c2s authentication."""
return str(getattr(settings, "PROSODY_AUTH_ENDPOINT", "http://127.0.0.1:8090/internal/prosody/auth/"))
return str(
getattr(
settings,
"PROSODY_AUTH_ENDPOINT",
"http://127.0.0.1:8090/internal/prosody/auth/",
)
)
def _recv_until(sock: socket.socket, patterns: list[bytes], timeout: float = 8.0, max_bytes: int = 16384) -> bytes:
def _recv_until(
sock: socket.socket,
patterns: list[bytes],
timeout: float = 8.0,
max_bytes: int = 16384,
) -> bytes:
"""Read from sock until one of the byte patterns appears or timeout/max_bytes hit."""
buf = b""
deadline = time.monotonic() + timeout
@@ -91,7 +103,9 @@ def _recv_until(sock: socket.socket, patterns: list[bytes], timeout: float = 8.0
return buf
def _component_handshake(address: str, port: int, jid: str, secret: str, timeout: float = 5.0) -> tuple[bool, str]:
def _component_handshake(
address: str, port: int, jid: str, secret: str, timeout: float = 5.0
) -> tuple[bool, str]:
"""
Attempt an XEP-0114 external component handshake.
@@ -123,7 +137,9 @@ def _component_handshake(address: str, port: int, jid: str, secret: str, timeout
token = hashlib.sha1((stream_id + secret).encode()).hexdigest()
sock.sendall(f"<handshake>{token}</handshake>".encode())
response = _recv_until(sock, [b"<handshake", b"<stream:error"], timeout=timeout)
response = _recv_until(
sock, [b"<handshake", b"<stream:error"], timeout=timeout
)
resp_text = response.decode(errors="replace")
if "<handshake/>" in resp_text or "<handshake />" in resp_text:
@@ -146,10 +162,11 @@ def _component_handshake(address: str, port: int, jid: str, secret: str, timeout
class _C2SResult:
"""Return value from _c2s_sasl_auth."""
def __init__(self, success: bool, stage: str, detail: str):
self.success = success # True = SASL <success/>
self.stage = stage # where we got to: tcp/starttls/tls/features/auth
self.detail = detail # human-readable explanation
self.success = success # True = SASL <success/>
self.stage = stage # where we got to: tcp/starttls/tls/features/auth
self.detail = detail # human-readable explanation
def __repr__(self):
return f"<C2SResult success={self.success} stage={self.stage!r} detail={self.detail!r}>"
@@ -178,8 +195,8 @@ def _c2s_sasl_auth(
9. Return _C2SResult with (True, "auth") on <success/> or (False, "auth") on <failure/>
"""
NS_STREAM = "http://etherx.jabber.org/streams"
NS_TLS = "urn:ietf:params:xml:ns:xmpp-tls"
NS_SASL = "urn:ietf:params:xml:ns:xmpp-sasl"
NS_TLS = "urn:ietf:params:xml:ns:xmpp-tls"
NS_SASL = "urn:ietf:params:xml:ns:xmpp-sasl"
def stream_open(to: str) -> bytes:
return (
@@ -201,19 +218,29 @@ def _c2s_sasl_auth(
raw.sendall(stream_open(domain))
# --- Receive pre-TLS features (expect <starttls>) ---
buf = _recv_until(raw, [b"</stream:features>", b"<stream:error"], timeout=timeout)
buf = _recv_until(
raw, [b"</stream:features>", b"<stream:error"], timeout=timeout
)
text = buf.decode(errors="replace")
if "<stream:error" in text:
return _C2SResult(False, "starttls", f"Stream error before features: {text[:200]}")
return _C2SResult(
False, "starttls", f"Stream error before features: {text[:200]}"
)
if "starttls" not in text.lower():
return _C2SResult(False, "starttls", f"No <starttls> in pre-TLS features: {text[:300]}")
return _C2SResult(
False, "starttls", f"No <starttls> in pre-TLS features: {text[:300]}"
)
# --- Negotiate STARTTLS ---
raw.sendall(f"<starttls xmlns='{NS_TLS}'/>".encode())
buf2 = _recv_until(raw, [b"<proceed", b"<failure"], timeout=timeout)
text2 = buf2.decode(errors="replace")
if "<proceed" not in text2:
return _C2SResult(False, "starttls", f"No <proceed/> after STARTTLS request: {text2[:200]}")
return _C2SResult(
False,
"starttls",
f"No <proceed/> after STARTTLS request: {text2[:200]}",
)
# --- Upgrade to TLS ---
ctx = ssl.create_default_context()
@@ -223,7 +250,9 @@ def _c2s_sasl_auth(
try:
tls = ctx.wrap_socket(raw, server_hostname=domain)
except ssl.SSLCertVerificationError as exc:
return _C2SResult(False, "tls", f"TLS cert verification failed for {domain!r}: {exc}")
return _C2SResult(
False, "tls", f"TLS cert verification failed for {domain!r}: {exc}"
)
except ssl.SSLError as exc:
return _C2SResult(False, "tls", f"TLS handshake error: {exc}")
@@ -231,23 +260,35 @@ def _c2s_sasl_auth(
# --- Re-open stream over TLS ---
tls.sendall(stream_open(domain))
buf3 = _recv_until(tls, [b"</stream:features>", b"<stream:error"], timeout=timeout)
buf3 = _recv_until(
tls, [b"</stream:features>", b"<stream:error"], timeout=timeout
)
text3 = buf3.decode(errors="replace")
if "<stream:error" in text3:
return _C2SResult(False, "features", f"Stream error after TLS: {text3[:200]}")
return _C2SResult(
False, "features", f"Stream error after TLS: {text3[:200]}"
)
mechanisms = re.findall(r"<mechanism>([^<]+)</mechanism>", text3, re.IGNORECASE)
if not mechanisms:
return _C2SResult(False, "features", f"No SASL mechanisms in post-TLS features: {text3[:300]}")
return _C2SResult(
False,
"features",
f"No SASL mechanisms in post-TLS features: {text3[:300]}",
)
if "PLAIN" not in [m.upper() for m in mechanisms]:
return _C2SResult(False, "features", f"SASL PLAIN not offered; got: {mechanisms}")
return _C2SResult(
False, "features", f"SASL PLAIN not offered; got: {mechanisms}"
)
# --- SASL PLAIN auth ---
credential = base64.b64encode(f"\x00{username}\x00{password}".encode()).decode()
tls.sendall(
f"<auth xmlns='{NS_SASL}' mechanism='PLAIN'>{credential}</auth>".encode()
)
buf4 = _recv_until(tls, [b"<success", b"<failure", b"<stream:error"], timeout=timeout)
buf4 = _recv_until(
tls, [b"<success", b"<failure", b"<stream:error"], timeout=timeout
)
text4 = buf4.decode(errors="replace")
if "<success" in text4:
@@ -256,7 +297,9 @@ def _c2s_sasl_auth(
# Extract the failure condition element name (e.g. not-authorized)
m = re.search(r"<failure[^>]*>\s*<([a-z-]+)", text4)
condition = m.group(1) if m else "unknown"
return _C2SResult(False, "auth", f"SASL PLAIN rejected: {condition} — {text4[:200]}")
return _C2SResult(
False, "auth", f"SASL PLAIN rejected: {condition} — {text4[:200]}"
)
if "<stream:error" in text4:
return _C2SResult(False, "auth", f"Stream error during auth: {text4[:200]}")
return _C2SResult(False, "auth", f"No auth response received: {text4[:200]}")
@@ -272,6 +315,7 @@ def _c2s_sasl_auth(
# Component tests (XEP-0114)
# ---------------------------------------------------------------------------
@unittest.skipUnless(_xmpp_configured(), "XMPP settings not configured")
class XMPPComponentTests(SimpleTestCase):
def test_component_port_reachable(self):
@@ -309,6 +353,7 @@ class XMPPComponentTests(SimpleTestCase):
# Auth bridge tests (what Prosody calls to validate user passwords)
# ---------------------------------------------------------------------------
@unittest.skipUnless(_xmpp_configured(), "XMPP settings not configured")
class XMPPAuthBridgeTests(SimpleTestCase):
"""
@@ -320,7 +365,12 @@ class XMPPAuthBridgeTests(SimpleTestCase):
def _parse_endpoint(self):
url = _prosody_auth_endpoint()
parsed = urllib.parse.urlparse(url)
return parsed.scheme, parsed.hostname, parsed.port or (443 if parsed.scheme == "https" else 80), parsed.path
return (
parsed.scheme,
parsed.hostname,
parsed.port or (443 if parsed.scheme == "https" else 80),
parsed.path,
)
def test_auth_endpoint_tcp_reachable(self):
"""Auth bridge port (8090) is listening inside the pod."""
@@ -349,13 +399,19 @@ class XMPPAuthBridgeTests(SimpleTestCase):
except (ConnectionRefusedError, OSError) as exc:
self.fail(f"Could not connect to auth bridge: {exc}")
# Should not return "1" (success) with wrong secret
self.assertNotEqual(body, "1", f"Auth bridge accepted a request with wrong secret (body={body!r})")
self.assertNotEqual(
body,
"1",
f"Auth bridge accepted a request with wrong secret (body={body!r})",
)
def test_auth_endpoint_isuser_returns_zero_or_one(self):
"""Auth bridge responds with '0' or '1' for an isuser query (not an error page)."""
secret = getattr(settings, "XMPP_SECRET", "")
_, host, port, path = self._parse_endpoint()
query = f"?command=isuser%3Anonexistent%3Azm.is&secret={urllib.parse.quote(secret)}"
query = (
f"?command=isuser%3Anonexistent%3Azm.is&secret={urllib.parse.quote(secret)}"
)
try:
conn = http.client.HTTPConnection(host, port, timeout=5)
conn.request("GET", path + query)
@@ -364,13 +420,18 @@ class XMPPAuthBridgeTests(SimpleTestCase):
conn.close()
except (ConnectionRefusedError, OSError) as exc:
self.fail(f"Could not connect to auth bridge: {exc}")
self.assertIn(body, ("0", "1"), f"Unexpected auth bridge response {body!r} (expected '0' or '1')")
self.assertIn(
body,
("0", "1"),
f"Unexpected auth bridge response {body!r} (expected '0' or '1')",
)
# ---------------------------------------------------------------------------
# c2s (client-to-server) tests — mirrors the phone's XMPP connection flow
# ---------------------------------------------------------------------------
@unittest.skipUnless(_xmpp_configured(), "XMPP settings not configured")
class XMPPClientAuthTests(SimpleTestCase):
"""
@@ -400,23 +461,26 @@ class XMPPClientAuthTests(SimpleTestCase):
port = _xmpp_c2s_port()
domain = _xmpp_domain()
result = _c2s_sasl_auth(
address=addr, port=port, domain=domain,
username="certcheck", password="certcheck",
verify_cert=True, timeout=10.0,
address=addr,
port=port,
domain=domain,
username="certcheck",
password="certcheck",
verify_cert=True,
timeout=10.0,
)
# We only care that we got past TLS — a SASL failure at stage "auth" is fine.
self.assertNotEqual(
result.stage, "tls",
result.stage,
"tls",
f"TLS cert validation failed for domain {domain!r}: {result.detail}\n"
"Phone will see a certificate error — it cannot connect at all."
"Phone will see a certificate error — it cannot connect at all.",
)
self.assertNotEqual(
result.stage, "tcp",
f"Could not reach c2s port at all: {result.detail}"
result.stage, "tcp", f"Could not reach c2s port at all: {result.detail}"
)
self.assertNotEqual(
result.stage, "starttls",
f"STARTTLS negotiation failed: {result.detail}"
result.stage, "starttls", f"STARTTLS negotiation failed: {result.detail}"
)
def test_c2s_sasl_plain_offered(self):
@@ -425,16 +489,21 @@ class XMPPClientAuthTests(SimpleTestCase):
port = _xmpp_c2s_port()
domain = _xmpp_domain()
result = _c2s_sasl_auth(
address=addr, port=port, domain=domain,
username="saslcheck", password="saslcheck",
verify_cert=False, timeout=10.0,
address=addr,
port=port,
domain=domain,
username="saslcheck",
password="saslcheck",
verify_cert=False,
timeout=10.0,
)
# We should reach the "auth" stage (SASL PLAIN was offered and we tried it).
# Reaching any earlier stage means SASL PLAIN wasn't offered or something broke.
self.assertIn(
result.stage, ("auth",),
result.stage,
("auth",),
f"Did not reach SASL auth stage — stopped at {result.stage!r}: {result.detail}\n"
"Check that allow_unencrypted_plain_auth = true in prosody config."
"Check that allow_unencrypted_plain_auth = true in prosody config.",
)
def test_c2s_invalid_credentials_rejected(self):
@@ -450,23 +519,31 @@ class XMPPClientAuthTests(SimpleTestCase):
port = _xmpp_c2s_port()
domain = _xmpp_domain()
result = _c2s_sasl_auth(
address=addr, port=port, domain=domain,
address=addr,
port=port,
domain=domain,
username="nobody_special",
password="definitely-wrong-password-xyz",
verify_cert=False, timeout=10.0,
verify_cert=False,
timeout=10.0,
)
self.assertFalse(
result.success,
f"Expected auth failure for invalid creds but got success: {result}",
)
self.assertFalse(result.success, f"Expected auth failure for invalid creds but got success: {result}")
self.assertEqual(
result.stage, "auth",
result.stage,
"auth",
f"Auth failed at stage {result.stage!r} (expected 'auth' / not-authorized).\n"
f"Detail: {result.detail}\n"
"This means Prosody cannot reach the Django auth bridge — "
"valid credentials would also fail. "
"Check that uWSGI has http-socket=127.0.0.1:8090 and the container is running."
"Check that uWSGI has http-socket=127.0.0.1:8090 and the container is running.",
)
self.assertIn(
"not-authorized", result.detail,
f"Expected 'not-authorized' failure, got: {result.detail}"
"not-authorized",
result.detail,
f"Expected 'not-authorized' failure, got: {result.detail}",
)
@unittest.skipUnless(
@@ -479,17 +556,22 @@ class XMPPClientAuthTests(SimpleTestCase):
Skipped unless env vars are set — run manually to verify end-to-end login.
"""
import os
addr = _xmpp_address()
port = _xmpp_c2s_port()
domain = _xmpp_domain()
username = os.environ["XMPP_TEST_USER"]
password = os.environ.get("XMPP_TEST_PASSWORD", "")
result = _c2s_sasl_auth(
address=addr, port=port, domain=domain,
username=username, password=password,
verify_cert=True, timeout=10.0,
address=addr,
port=port,
domain=domain,
username=username,
password=password,
verify_cert=True,
timeout=10.0,
)
self.assertTrue(
result.success,
f"Login with XMPP_TEST_USER={username!r} failed at stage {result.stage!r}: {result.detail}"
f"Login with XMPP_TEST_USER={username!r} failed at stage {result.stage!r}: {result.detail}",
)