lint

tests/unittests/test_third_party_http_functions.py

@@ -134,14 +134,14 @@ def test_raw_body_bytes(self):
             with open(image_file, 'rb') as image:
                 img = image.read()
                 sanitized_image = urllib.parse.quote(img)
-                sanitized_img_len = len(img)
+                sanitized_img_len = len(sanitized_image)
                 r = self.webhost.request('POST', 'raw_body_bytes', data=img,
                                          no_prefix=True)
 
             received_body_len = int(r.headers['body-len'])
             self.assertEqual(received_body_len, sanitized_img_len)
 
-            body = r.content
+            body = urllib.parse.unquote_to_bytes(r.content)
             try:
                 received_img_file = parent_dir / 'received_img.png'
                 with open(received_img_file, 'wb') as received_img:

tests/unittests/third_party_http_functions/stein/asgi_function/function_app.py

@@ -1,8 +1,9 @@
 import asyncio
 import logging
+import re
 import sys
 from urllib.request import urlopen
-from urllib.parse import urlparse
+import urllib.parse
 
 import azure.functions as func
 from fastapi import FastAPI, Request, Response
@@ -132,7 +133,9 @@ async def print_logging(message: str = "", flush: str = 'false',
 @fast_app.post("/raw_body_bytes")
 async def raw_body_bytes(request: Request):
     raw_body = await request.body()
-    return Response(content=raw_body, headers={'body-len': str(len(raw_body))})
+    sanitized_body = urllib.parse.quote(raw_body)
+    return Response(content=sanitized_body,
+                    headers={'body-len': str(len(sanitized_body))})
 
 
 @fast_app.get("/return_http_no_body")
@@ -147,10 +150,17 @@ async def return_http(request: Request):
 
 @fast_app.get("/return_http_redirect")
 async def return_http_redirect(request: Request, code: str = ''):
+    allowed_url_pattern = r"^http://.+"
+
     location = 'return_http?code={}'.format(code)
     redirect_url = f"http://{request.url.components[1]}/{location}"
+    if re.match(allowed_url_pattern, redirect_url):
+        # Redirect URL is in the expected format
+        return RedirectResponse(status_code=302,
+                                url=redirect_url)
+    # Redirect URL was not in the expected format
     return RedirectResponse(status_code=302,
-                            url=redirect_url)
+                            url='/')
 
 
 @fast_app.get("/unhandled_error")

tests/unittests/third_party_http_functions/stein/wsgi_function/function_app.py

@@ -62,8 +62,8 @@ def print_logging():
 def raw_body_bytes():
     body = request.get_data()
 
-    #sanitized_body = urllib.parse.quote(body)
-    return Response(body, headers={'body-len': str(len(body))})
+    sanitized_body = urllib.parse.quote(body)
+    return Response(sanitized_body, headers={'body-len': str(len(sanitized_body))})
 
 
 @flask_app.get("/return_http_no_body")