paulmataruso/zulip
1
0
Fork 0
mirror of https://github.com/zulip/zulip.git synced 2025-11-03 21:43:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

zephyr: Switch from py3dns to dnspython.

Browse Source 
Signed-off-by: Anders Kaseorg <anders@zulip.com>
This commit is contained in:
Anders Kaseorg
2025-02-25 10:31:44 -08:00
committed by Tim Abbott
parent 0338fd7357
commit 1f085a920a
7 changed files with 83 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
zerver/forms.py
View File

@@ -3,7 +3,7 @@ import re
from email.headerregistry import Address from email.headerregistry import Address
from typing import Any from typing import Any
import DNS import dns.resolver
from django import forms from django import forms
from django.conf import settings from django.conf import settings
from django.contrib.auth import authenticate, password_validation from django.contrib.auth import authenticate, password_validation
@@ -66,14 +66,11 @@ def email_is_not_mit_mailing_list(email: str) -> None:
if address.domain == "mit.edu": if address.domain == "mit.edu":
# Check whether the user exists and can get mail. # Check whether the user exists and can get mail.
try: try:
DNS.dnslookup(f"{address.username}.pobox.ns.athena.mit.edu", DNS.Type.TXT) dns.resolver.resolve(f"{address.username}.pobox.ns.athena.mit.edu", "TXT")
except DNS.Base.ServerError as e: except dns.resolver.NXDOMAIN:
if e.rcode == DNS.Status.NXDOMAIN: # This error is Markup only because 1. it needs to render HTML
# This error is Markup only because 1. it needs to render HTML # 2. It's not formatted with any user input.
# 2. It's not formatted with any user input. raise ValidationError(MIT_VALIDATION_ERROR)
raise ValidationError(MIT_VALIDATION_ERROR)
else:
raise AssertionError("Unexpected DNS error")
class OverridableValidationError(ValidationError): class OverridableValidationError(ValidationError):

15
zerver/lib/test_helpers.py
View File

@@ -12,6 +12,8 @@ from unittest import mock
from unittest.mock import patch from unittest.mock import patch
import boto3.session import boto3.session
import dns.rdtypes.ANY.TXT
import dns.resolver
import fakeldap import fakeldap
import ldap import ldap
import orjson import orjson
@@ -793,3 +795,16 @@ def ratelimit_rule(
def consume_response(response: HttpResponseBase) -> None: def consume_response(response: HttpResponseBase) -> None:
assert response.streaming assert response.streaming
collections.deque(response, maxlen=0) collections.deque(response, maxlen=0)
def dns_txt_answer(name_str: str, txt: str) -> dns.resolver.Answer:
name = dns.name.from_text(name_str)
rdclass = dns.rdataclass.IN
rdtype = dns.rdatatype.TXT
response = dns.message.make_query(
name, rdtype, rdclass, flags=dns.flags.QR | dns.flags.RA | dns.flags.RD
)
response.find_rrset(dns.message.ANSWER, name, rdclass, rdtype, create=True).add(
dns.rdtypes.ANY.TXT.TXT(rdclass, rdtype, txt)
)
return dns.resolver.Answer(name, rdtype, rdclass, response)

8
zerver/lib/zephyr.py
View File

@@ -1,7 +1,7 @@
import re import re
import traceback import traceback
import DNS import dns.resolver
def compute_mit_user_fullname(email: str) -> str: def compute_mit_user_fullname(email: str) -> str:
@@ -9,13 +9,13 @@ def compute_mit_user_fullname(email: str) -> str:
# Input is either e.g. username@mit.edu or user|CROSSREALM.INVALID@mit.edu # Input is either e.g. username@mit.edu or user|CROSSREALM.INVALID@mit.edu
match_user = re.match(r"^([a-zA-Z0-9_.-]+)(\|.+)?@mit\.edu$", email.lower()) match_user = re.match(r"^([a-zA-Z0-9_.-]+)(\|.+)?@mit\.edu$", email.lower())
if match_user and match_user.group(2) is None: if match_user and match_user.group(2) is None:
answer = DNS.dnslookup(f"{match_user.group(1)}.passwd.ns.athena.mit.edu", DNS.Type.TXT) answer = dns.resolver.resolve(f"{match_user.group(1)}.passwd.ns.athena.mit.edu", "TXT")
hesiod_name = answer[0][0].decode().split(":")[4].split(",")[0].strip() hesiod_name = answer[0].strings[0].decode().split(":")[4].split(",")[0].strip()
if hesiod_name != "": if hesiod_name != "":
return hesiod_name return hesiod_name
elif match_user: elif match_user:
return match_user.group(1).lower() + "@" + match_user.group(2).upper().removeprefix("|") return match_user.group(1).lower() + "@" + match_user.group(2).upper().removeprefix("|")
except DNS.Base.ServerError: except dns.resolver.NXDOMAIN:
pass pass
except Exception: except Exception:
print(f"Error getting fullname for {email}:") print(f"Error getting fullname for {email}:")

43
zerver/tests/test_external.py
View File

@@ -5,7 +5,7 @@ from contextlib import contextmanager
from typing import IO, TYPE_CHECKING, Any from typing import IO, TYPE_CHECKING, Any
from unittest import mock, skipUnless from unittest import mock, skipUnless
import DNS import dns.resolver
import orjson import orjson
from circuitbreaker import CircuitBreakerMonitor from circuitbreaker import CircuitBreakerMonitor
from django.conf import settings from django.conf import settings
@@ -23,7 +23,7 @@ from zerver.lib.rate_limiter import (
get_tor_ips, get_tor_ips,
) )
from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import ratelimit_rule from zerver.lib.test_helpers import dns_txt_answer, ratelimit_rule
from zerver.lib.zephyr import compute_mit_user_fullname from zerver.lib.zephyr import compute_mit_user_fullname
from zerver.models import PushDeviceToken, UserProfile from zerver.models import PushDeviceToken, UserProfile
@@ -37,43 +37,44 @@ if TYPE_CHECKING:
class MITNameTest(ZulipTestCase): class MITNameTest(ZulipTestCase):
def test_valid_hesiod(self) -> None: def test_valid_hesiod(self) -> None:
with mock.patch( with mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[ return_value=dns_txt_answer(
[b"starnine:*:84233:101:Athena Consulting Exchange User,,,:/mit/starnine:/bin/bash"] "starnine.passwd.ns.athena.mit.edu.",
], "starnine:*:84233:101:Athena Consulting Exchange User,,,:/mit/starnine:/bin/bash",
),
): ):
self.assertEqual( self.assertEqual(
compute_mit_user_fullname(self.mit_email("starnine")), compute_mit_user_fullname(self.mit_email("starnine")),
"Athena Consulting Exchange User", "Athena Consulting Exchange User",
) )
with mock.patch( with mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"sipbexch:*:87824:101:Exch Sipb,,,:/mit/sipbexch:/bin/athena/bash"]], return_value=dns_txt_answer(
"sipbexch.passwd.ns.athena.mit.edu.",
"sipbexch:*:87824:101:Exch Sipb,,,:/mit/sipbexch:/bin/athena/bash",
),
): ):
self.assertEqual(compute_mit_user_fullname("sipbexch@mit.edu"), "Exch Sipb") self.assertEqual(compute_mit_user_fullname("sipbexch@mit.edu"), "Exch Sipb")
def test_invalid_hesiod(self) -> None: def test_invalid_hesiod(self) -> None:
with mock.patch( with mock.patch("dns.resolver.resolve", side_effect=dns.resolver.NXDOMAIN):
"DNS.dnslookup", side_effect=DNS.Base.ServerError("DNS query status: NXDOMAIN", 3)
):
self.assertEqual(compute_mit_user_fullname("1234567890@mit.edu"), "1234567890@mit.edu") self.assertEqual(compute_mit_user_fullname("1234567890@mit.edu"), "1234567890@mit.edu")
with mock.patch( with mock.patch("dns.resolver.resolve", side_effect=dns.resolver.NXDOMAIN):
"DNS.dnslookup", side_effect=DNS.Base.ServerError("DNS query status: NXDOMAIN", 3)
):
self.assertEqual(compute_mit_user_fullname("ec-discuss@mit.edu"), "ec-discuss@mit.edu") self.assertEqual(compute_mit_user_fullname("ec-discuss@mit.edu"), "ec-discuss@mit.edu")
def test_mailinglist(self) -> None: def test_mailinglist(self) -> None:
with mock.patch( with mock.patch("dns.resolver.resolve", side_effect=dns.resolver.NXDOMAIN):
"DNS.dnslookup", side_effect=DNS.Base.ServerError("DNS query status: NXDOMAIN", 3)
):
self.assertRaises(ValidationError, email_is_not_mit_mailing_list, "1234567890@mit.edu") self.assertRaises(ValidationError, email_is_not_mit_mailing_list, "1234567890@mit.edu")
with mock.patch( with mock.patch("dns.resolver.resolve", side_effect=dns.resolver.NXDOMAIN):
"DNS.dnslookup", side_effect=DNS.Base.ServerError("DNS query status: NXDOMAIN", 3)
):
self.assertRaises(ValidationError, email_is_not_mit_mailing_list, "ec-discuss@mit.edu") self.assertRaises(ValidationError, email_is_not_mit_mailing_list, "ec-discuss@mit.edu")
def test_notmailinglist(self) -> None: def test_notmailinglist(self) -> None:
with mock.patch("DNS.dnslookup", return_value=[[b"POP IMAP.EXCHANGE.MIT.EDU starnine"]]): with mock.patch(
"dns.resolver.resolve",
return_value=dns_txt_answer(
"starnine.pobox.ns.athena.mit.edu.", "POP IMAP.EXCHANGE.MIT.EDU starnine"
),
):
email_is_not_mit_mailing_list("sipbexch@mit.edu") email_is_not_mit_mailing_list("sipbexch@mit.edu")

17
zerver/tests/test_message_send.py
View File

@@ -50,6 +50,7 @@ from zerver.lib.per_request_cache import flush_per_request_caches
from zerver.lib.streams import create_stream_if_needed from zerver.lib.streams import create_stream_if_needed
from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import ( from zerver.lib.test_helpers import (
dns_txt_answer,
get_user_messages, get_user_messages,
make_client, make_client,
message_stream_count, message_stream_count,
@@ -1052,10 +1053,11 @@ class MessagePOSTTest(ZulipTestCase):
} }
with mock.patch( with mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[ return_value=dns_txt_answer(
[b"starnine:*:84233:101:Athena Consulting Exchange User,,,:/mit/starnine:/bin/bash"] "starnine.passwd.ns.athena.mit.edu.",
], "starnine:*:84233:101:Athena Consulting Exchange User,,,:/mit/starnine:/bin/bash",
),
): ):
result1 = self.api_post( result1 = self.api_post(
self.mit_user("starnine"), "/api/v1/messages", msg, subdomain="zephyr" self.mit_user("starnine"), "/api/v1/messages", msg, subdomain="zephyr"
@@ -1063,8 +1065,11 @@ class MessagePOSTTest(ZulipTestCase):
self.assert_json_success(result1) self.assert_json_success(result1)
with mock.patch( with mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"espuser:*:95494:101:Esp Classroom,,,:/mit/espuser:/bin/athena/bash"]], return_value=dns_txt_answer(
("espuser.passwd.ns.athena.mit.edu."),
"espuser:*:95494:101:Esp Classroom,,,:/mit/espuser:/bin/athena/bash",
),
): ):
result2 = self.api_post( result2 = self.api_post(
self.mit_user("espuser"), "/api/v1/messages", msg, subdomain="zephyr" self.mit_user("espuser"), "/api/v1/messages", msg, subdomain="zephyr"

19
zerver/tests/test_mirror_users.py
View File

@@ -7,7 +7,10 @@ from django.utils.timezone import now as timezone_now
from zerver.actions.message_send import create_mirror_user_if_needed from zerver.actions.message_send import create_mirror_user_if_needed
from zerver.lib.create_user import create_user_profile from zerver.lib.create_user import create_user_profile
from zerver.lib.test_classes import ZulipTestCase from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import reset_email_visibility_to_everyone_in_zulip_realm from zerver.lib.test_helpers import (
dns_txt_answer,
reset_email_visibility_to_everyone_in_zulip_realm,
)
from zerver.models import UserProfile from zerver.models import UserProfile
from zerver.models.clients import get_client from zerver.models.clients import get_client
from zerver.models.realms import get_realm from zerver.models.realms import get_realm
@@ -49,8 +52,11 @@ class MirroredMessageUsersTest(ZulipTestCase):
) )
@mock.patch( @mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh"]], return_value=dns_txt_answer(
"sipbtest.passwd.ns.athena.mit.edu.",
"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh",
),
) )
def test_zephyr_mirror_new_recipient(self, ignored: object) -> None: def test_zephyr_mirror_new_recipient(self, ignored: object) -> None:
"""Test mirror dummy user creation for direct message recipients""" """Test mirror dummy user creation for direct message recipients"""
@@ -79,8 +85,11 @@ class MirroredMessageUsersTest(ZulipTestCase):
self.assertTrue(bob.is_mirror_dummy) self.assertTrue(bob.is_mirror_dummy)
@mock.patch( @mock.patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh"]], return_value=dns_txt_answer(
"sipbtest.passwd.ns.athena.mit.edu.",
"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh",
),
) )
def test_zephyr_mirror_new_sender(self, ignored: object) -> None: def test_zephyr_mirror_new_sender(self, ignored: object) -> None:
"""Test mirror dummy user creation for sender when sending to stream""" """Test mirror dummy user creation for sender when sending to stream"""

15
zerver/tests/test_signup.py
View File

@@ -53,6 +53,7 @@ from zerver.lib.test_classes import ZulipTestCase
from zerver.lib.test_helpers import ( from zerver.lib.test_helpers import (
HostRequestMock, HostRequestMock,
avatar_disk_path, avatar_disk_path,
dns_txt_answer,
find_key_by_email, find_key_by_email,
get_test_image_file, get_test_image_file,
load_subdomain_token, load_subdomain_token,
@@ -4135,8 +4136,11 @@ class UserSignUpTest(ZulipTestCase):
self.assertEqual(mirror_dummy.role, UserProfile.ROLE_GUEST) self.assertEqual(mirror_dummy.role, UserProfile.ROLE_GUEST)
@patch( @patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh"]], return_value=dns_txt_answer(
"sipbtest.passwd.ns.athena.mit.edu.",
"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh",
),
) )
def test_registration_of_mirror_dummy_user(self, ignored: Any) -> None: def test_registration_of_mirror_dummy_user(self, ignored: Any) -> None:
password = "test" password = "test"
@@ -4216,8 +4220,11 @@ class UserSignUpTest(ZulipTestCase):
self.assert_logged_in_user_id(user_profile.id) self.assert_logged_in_user_id(user_profile.id)
@patch( @patch(
"DNS.dnslookup", "dns.resolver.resolve",
return_value=[[b"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh"]], return_value=dns_txt_answer(
"sipbtest.passwd.ns.athena.mit.edu.",
"sipbtest:*:20922:101:Fred Sipb,,,:/mit/sipbtest:/bin/athena/tcsh",
),
) )
def test_registration_of_active_mirror_dummy_user(self, ignored: Any) -> None: def test_registration_of_active_mirror_dummy_user(self, ignored: Any) -> None:
""" """