Merge pull request #24 from danpilgrim-aws/master

feat: add content security policy recommendations to requests without csp
aws · Oct 28, 2022 · a1ba1a7 · a1ba1a7
2 parents 2f515c2 + e9e1500
commit a1ba1a7
Show file tree

Hide file tree

Showing 3 changed files with 75 additions and 1 deletion.
diff --git a/aws_jupyter_proxy/awsproxy.py b/aws_jupyter_proxy/awsproxy.py
@@ -99,6 +99,10 @@ def _finish_response(self, response: HTTPResponse):
             if self._is_blacklisted_response_header(name, value):
                 continue
             self.set_header(name, value)
+        csp_value = response.headers.get(
+            "Content-Security-Policy", "upgrade-insecure-requests; base-uri 'none';"
+        )
+        self.set_header("Content-Security-Policy", csp_value)
         super(APIHandler, self).finish(response.body or None)
 
     async def post(self, *args):

diff --git a/setup.py b/setup.py
@@ -5,7 +5,7 @@
 
 setuptools.setup(
     name="aws_jupyter_proxy",
-    version="0.3.3",
+    version="0.3.4",
     url="https://github.com/aws/aws-jupyter-proxy",
     author="Amazon Web Services",
     description="A Jupyter server extension to proxy requests with AWS SigV4 authentication",

diff --git a/tests/unit/test_handlers.py b/tests/unit/test_handlers.py
@@ -78,5 +78,75 @@ def test_downstream_success_blacklisted_headers_removed(self, mock_awsproxy):
         assert "foo-abc" == response.headers["X-Amz-RequestId"]
         assert "awsservice.amazonaws.com" == response.headers["Host"]
 
+    @patch("aws_jupyter_proxy.awsproxy.AwsProxyRequest")
+    def test_downstream_success_with_content_security_policy(self, mock_awsproxy):
+        # Given
+        mock_execute_downstream = CoroutineMock()
+        mock_execute_downstream.return_value = HTTPResponse(
+            request=HTTPRequest(url="https://awsservice.amazonaws.com/"),
+            code=200,
+            headers=HTTPHeaders(
+                {
+                    "Host": "awsservice.amazonaws.com",
+                    "X-Amz-RequestId": "foo-abc",
+                    "Transfer-Encoding": "chunked",
+                    "Content-Security-Policy": "default-src 'none';",
+                }
+            ),
+            buffer=BytesIO(b"SomeResponse"),
+        )
+
+        mock_instance = mock_awsproxy.return_value
+        mock_instance.execute_downstream = mock_execute_downstream
+
+        # When
+        response = self.fetch("/awsproxy")
+
+        # Then
+        mock_execute_downstream.assert_awaited_once()
+        assert 200 == response.code
+        assert b"SomeResponse" == response.body
+        assert "Transfer-Encoding" not in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert "default-src 'none';" == response.headers["Content-Security-Policy"]
+        assert "foo-abc" == response.headers["X-Amz-RequestId"]
+        assert "awsservice.amazonaws.com" == response.headers["Host"]
+
+    @patch("aws_jupyter_proxy.awsproxy.AwsProxyRequest")
+    def test_downstream_success_without_content_security_policy(self, mock_awsproxy):
+        # Given
+        mock_execute_downstream = CoroutineMock()
+        mock_execute_downstream.return_value = HTTPResponse(
+            request=HTTPRequest(url="https://awsservice.amazonaws.com/"),
+            code=200,
+            headers=HTTPHeaders(
+                {
+                    "Host": "awsservice.amazonaws.com",
+                    "X-Amz-RequestId": "foo-abc",
+                    "Transfer-Encoding": "chunked",
+                }
+            ),
+            buffer=BytesIO(b"SomeResponse"),
+        )
+
+        mock_instance = mock_awsproxy.return_value
+        mock_instance.execute_downstream = mock_execute_downstream
+
+        # When
+        response = self.fetch("/awsproxy")
+
+        # Then
+        mock_execute_downstream.assert_awaited_once()
+        assert 200 == response.code
+        assert b"SomeResponse" == response.body
+        assert "Transfer-Encoding" not in response.headers
+        assert "Content-Security-Policy" in response.headers
+        assert (
+            "upgrade-insecure-requests; base-uri 'none';"
+            == response.headers["Content-Security-Policy"]
+        )
+        assert "foo-abc" == response.headers["X-Amz-RequestId"]
+        assert "awsservice.amazonaws.com" == response.headers["Host"]
+
     def get_app(self):
         return tornado.web.Application(awsproxy_handlers)