url_encoding: Use proper parsing for query string appending.

Signed-off-by: Anders Kaseorg <anders@zulip.com>

zerver/lib/url_encoding.py

@@ -1,7 +1,6 @@
-import urllib
 from typing import Any, Dict, List
+from urllib.parse import quote, urlsplit
 
-from zerver.lib.pysa import mark_sanitized
 from zerver.lib.topic import get_topic_from_message_info
 from zerver.models import Realm, Stream, UserProfile
 
@@ -10,7 +9,7 @@ def hash_util_encode(string: str) -> str:
     # Do the same encoding operation as hash_util.encodeHashComponent on the
     # frontend.
     # `safe` has a default value of "/", but we want those encoded, too.
-    return urllib.parse.quote(string, safe=b"").replace(".", "%2E").replace("%", ".")
+    return quote(string, safe=b"").replace(".", "%2E").replace("%", ".")
 
 
 def encode_stream(stream_id: int, stream_name: str) -> str:
@@ -100,14 +99,7 @@ def near_pm_message_url(realm: Realm, message: Dict[str, Any]) -> str:
     return full_url
 
 
-def add_query_to_redirect_url(original_url: str, query: str) -> str:
-    # Using 'mark_sanitized' because user-controlled data after the '?' is
-    # not relevant for open redirects
-    return original_url + "?" + mark_sanitized(query)
-
-
-def add_query_arg_to_redirect_url(original_url: str, query_arg: str) -> str:
-    assert "?" in original_url
-    # Using 'mark_sanitized' because user-controlled data after the '?' is
-    # not relevant for open redirects
-    return original_url + "&" + mark_sanitized(query_arg)
+def append_url_query_string(original_url: str, query: str) -> str:
+    u = urlsplit(original_url)
+    query = u.query + ("&" if u.query and query else "") + query
+    return u._replace(query=query).geturl()

zerver/views/auth.py

@@ -63,7 +63,7 @@ from zerver.lib.response import json_success
 from zerver.lib.sessions import set_expirable_session_var
 from zerver.lib.subdomains import get_subdomain, is_subdomain_root_or_alias
 from zerver.lib.types import ViewFuncT
-from zerver.lib.url_encoding import add_query_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.user_agent import parse_user_agent
 from zerver.lib.users import get_api_key
 from zerver.lib.utils import has_api_key_format
@@ -385,9 +385,7 @@ def create_response_for_otp_flow(
     }
     # We can't use HttpResponseRedirect, since it only allows HTTP(S) URLs
     response = HttpResponse(status=302)
-    response["Location"] = add_query_to_redirect_url(
-        "zulip://login", urllib.parse.urlencode(params)
-    )
+    response["Location"] = append_url_query_string("zulip://login", urllib.parse.urlencode(params))
 
     return response
 
@@ -530,7 +528,7 @@ def oauth_redirect_to_root(
 
     params = {**params, **extra_url_params}
 
-    return redirect(add_query_to_redirect_url(main_site_uri, urllib.parse.urlencode(params)))
+    return redirect(append_url_query_string(main_site_uri, urllib.parse.urlencode(params)))
 
 
 def handle_desktop_flow(func: ViewFuncT) -> ViewFuncT:
@@ -554,7 +552,7 @@ def start_remote_user_sso(request: HttpRequest) -> HttpResponse:
     to do authentication, so we need this additional endpoint.
     """
     query = request.META["QUERY_STRING"]
-    return redirect(add_query_to_redirect_url(reverse(remote_user_sso), query))
+    return redirect(append_url_query_string(reverse(remote_user_sso), query))
 
 
 @handle_desktop_flow
@@ -759,7 +757,7 @@ def login_page(
     if is_subdomain_root_or_alias(request) and settings.ROOT_DOMAIN_LANDING_PAGE:
         redirect_url = reverse("realm_redirect")
         if request.GET:
-            redirect_url = add_query_to_redirect_url(redirect_url, request.GET.urlencode())
+            redirect_url = append_url_query_string(redirect_url, request.GET.urlencode())
         return HttpResponseRedirect(redirect_url)
 
     realm = get_realm_from_request(request)
@@ -989,7 +987,7 @@ def logout_then_login(request: HttpRequest, **kwargs: Any) -> HttpResponse:
 
 def password_reset(request: HttpRequest) -> HttpResponse:
     if is_subdomain_root_or_alias(request) and settings.ROOT_DOMAIN_LANDING_PAGE:
-        redirect_url = add_query_to_redirect_url(
+        redirect_url = append_url_query_string(
             reverse("realm_redirect"), urlencode({"next": reverse("password_reset")})
         )
         return HttpResponseRedirect(redirect_url)

zerver/views/realm_icon.py

@@ -9,7 +9,7 @@ from zerver.lib.exceptions import JsonableError
 from zerver.lib.realm_icon import realm_icon_url
 from zerver.lib.response import json_success
 from zerver.lib.upload import upload_icon_image
-from zerver.lib.url_encoding import add_query_arg_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.models import UserProfile
 
 
@@ -60,5 +60,5 @@ def get_icon_backend(request: HttpRequest, user_profile: UserProfile) -> HttpRes
     # our templates depend on being able to use the ampersand to
     # add query parameters to our url, get_icon_url does '?version=version_number'
     # hacks to prevent us from having to jump through decode/encode hoops.
-    url = add_query_arg_to_redirect_url(url, request.META["QUERY_STRING"])
+    url = append_url_query_string(url, request.META["QUERY_STRING"])
     return redirect(url)

zerver/views/realm_logo.py

@@ -10,7 +10,7 @@ from zerver.lib.realm_logo import get_realm_logo_url
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
 from zerver.lib.upload import upload_logo_image
-from zerver.lib.url_encoding import add_query_arg_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.validator import check_bool
 from zerver.models import UserProfile
 
@@ -62,5 +62,5 @@ def get_logo_backend(
     # our templates depend on being able to use the ampersand to
     # add query parameters to our url, get_logo_url does '?version=version_number'
     # hacks to prevent us from having to jump through decode/encode hoops.
-    url = add_query_arg_to_redirect_url(url, request.META["QUERY_STRING"])
+    url = append_url_query_string(url, request.META["QUERY_STRING"])
     return redirect(url)

zerver/views/registration.py

@@ -51,7 +51,7 @@ from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.send_email import EmailNotDeliveredException, FromAddress, send_email
 from zerver.lib.sessions import get_expirable_session_var
 from zerver.lib.subdomains import get_subdomain, is_root_domain_available
-from zerver.lib.url_encoding import add_query_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.users import get_accounts_for_email
 from zerver.lib.validator import to_converted_or_fallback, to_non_negative_int, to_timezone_or_empty
 from zerver.lib.zephyr import compute_mit_user_fullname
@@ -405,7 +405,7 @@ def accounts_register(
                     # is hidden for most users.
                     view_url = reverse("login")
                     query = urlencode({"email": email})
-                    redirect_url = add_query_to_redirect_url(view_url, query)
+                    redirect_url = append_url_query_string(view_url, query)
                     return HttpResponseRedirect(redirect_url)
             elif not realm_creation:
                 # Since we'll have created a user, we now just log them in.
@@ -574,7 +574,7 @@ def send_confirm_registration_email(
 
 def redirect_to_email_login_url(email: str) -> HttpResponseRedirect:
     login_url = reverse("login")
-    redirect_url = add_query_to_redirect_url(
+    redirect_url = append_url_query_string(
         login_url, urlencode({"email": email, "already_registered": 1})
     )
     return HttpResponseRedirect(redirect_url)
@@ -775,7 +775,7 @@ def find_account(
             # feature can be used to ascertain which email addresses
             # are associated with Zulip.
             data = urllib.parse.urlencode({"emails": ",".join(emails)})
-            return redirect(add_query_to_redirect_url(url, data))
+            return redirect(append_url_query_string(url, data))
     else:
         form = FindMyTeamForm()
         # The below validation is perhaps unnecessary, in that we

zerver/views/users.py

@@ -40,7 +40,7 @@ from zerver.lib.response import json_success
 from zerver.lib.streams import access_stream_by_id, access_stream_by_name, subscribed_to_stream
 from zerver.lib.types import ProfileDataElementValue, Validator
 from zerver.lib.upload import upload_avatar_image
-from zerver.lib.url_encoding import add_query_arg_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.users import (
     access_bot_by_id,
     access_user_by_id,
@@ -249,7 +249,7 @@ def avatar(
     # add query parameters to our url, get_avatar_url does '?x=x'
     # hacks to prevent us from having to jump through decode/encode hoops.
     assert url is not None
-    url = add_query_arg_to_redirect_url(url, request.META["QUERY_STRING"])
+    url = append_url_query_string(url, request.META["QUERY_STRING"])
     return redirect(url)
 
 

zerver/views/video_calls.py

@@ -29,7 +29,7 @@ from zerver.lib.pysa import mark_sanitized
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
 from zerver.lib.subdomains import get_subdomain
-from zerver.lib.url_encoding import add_query_arg_to_redirect_url, add_query_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.validator import check_dict, check_string
 from zerver.models import UserProfile, get_realm
 
@@ -191,7 +191,7 @@ def get_bigbluebutton_url(request: HttpRequest, user_profile: UserProfile) -> Ht
             + settings.BIG_BLUE_BUTTON_SECRET
         ).encode()
     ).hexdigest()
-    url = add_query_to_redirect_url(
+    url = append_url_query_string(
         "/calls/bigbluebutton/join",
         urlencode(
             {
@@ -225,7 +225,7 @@ def join_bigbluebutton(
     else:
         try:
             response = VideoCallSession().get(
-                add_query_to_redirect_url(
+                append_url_query_string(
                     settings.BIG_BLUE_BUTTON_URL + "api/create",
                     urlencode(
                         {
@@ -260,7 +260,7 @@ def join_bigbluebutton(
         checksum = hashlib.sha1(
             ("join" + join_params + settings.BIG_BLUE_BUTTON_SECRET).encode()
         ).hexdigest()
-        redirect_url_base = add_query_to_redirect_url(
+        redirect_url_base = append_url_query_string(
             settings.BIG_BLUE_BUTTON_URL + "api/join", join_params
         )
-        return redirect(add_query_arg_to_redirect_url(redirect_url_base, "checksum=" + checksum))
+        return redirect(append_url_query_string(redirect_url_base, "checksum=" + checksum))

zproject/backends.py

@@ -75,7 +75,7 @@ from zerver.lib.redis_utils import get_dict_from_redis, get_redis_client, put_di
 from zerver.lib.request import RequestNotes
 from zerver.lib.subdomains import get_subdomain
 from zerver.lib.types import ProfileDataElementValue
-from zerver.lib.url_encoding import add_query_to_redirect_url
+from zerver.lib.url_encoding import append_url_query_string
 from zerver.lib.users import check_full_name, validate_user_custom_profile_field
 from zerver.models import (
     CustomProfileField,
@@ -1429,7 +1429,7 @@ def redirect_deactivated_user_to_login(realm: Realm, email: str) -> HttpResponse
     # Specifying the template name makes sure that the user is not redirected to dev_login in case of
     # a deactivated account on a test server.
     login_url = reverse("login_page", kwargs={"template_name": "zerver/login.html"})
-    redirect_url = add_query_to_redirect_url(
+    redirect_url = append_url_query_string(
         realm.uri + login_url, urlencode({"is_deactivated": email})
     )
     return HttpResponseRedirect(redirect_url)