Add check for same-origin requests for unsafe methods

Author: takluyver

tornado/test/web_test.py

@@ -3130,6 +3130,45 @@ def test_xsrf_httponly(self):
         self.assertTrue(abs((expires - header_expires).total_seconds()) < 10)
 
 
+class CheckSameOriginTest(SimpleHandlerTestCase):
+    class Handler(RequestHandler):
+        def post(self):
+            self.write("ok")
+
+    def get_app_kwargs(self):
+        return dict(check_same_origin=True)
+
+    def _post(self, headers):
+        return self.fetch("/", method="POST", body="x=1", headers=headers)
+
+    def test_sec_fetch_site_success(self):
+        response = self._post({"Sec-Fetch-Site": "same-origin"})
+        self.assertEqual(response.code, 200)
+
+    def test_sec_fetch_site_fail(self):
+        with ExpectLog(gen_log, ".*Cross-origin request"):
+            response = self._post({"Sec-Fetch-Site": "cross-site"})
+        self.assertEqual(response.code, 403)
+
+    def test_fallback_success(self):
+        response = self._post({"Origin": self.get_url("")})
+        self.assertEqual(response.code, 200)
+
+    def test_fallback_referrer_success(self):
+        response = self._post({"Referrer": self.get_url("/foo/bar")})
+        self.assertEqual(response.code, 200)
+
+    def test_fallback_fail(self):
+        with ExpectLog(gen_log, ".*Cross-origin request"):
+            response = self._post({"Origin": "https://evil.example.com/"})
+        self.assertEqual(response.code, 403)
+
+    def test_fallback_no_origin(self):
+        with ExpectLog(gen_log, ".*No Origin/Referrer"):
+            response = self._post({})
+        self.assertEqual(response.code, 403)
+
+
 class FinishExceptionTest(SimpleHandlerTestCase):
     class Handler(RequestHandler):
         def get(self):

tornado/web.py

@@ -1690,6 +1690,25 @@ def xsrf_form_html(self) -> str:
             + '"/>'
         )
 
+    def check_same_origin(self) -> None:
+        """Verify that non-safe methods come from a same-origin request"""
+        headers = self.request.headers
+        if (sfs := headers.get("Sec-Fetch-Site")) is not None:
+            # All major browsers send the Sec-Fetch-Site header since ~2023
+            # for 'potentially trustworthy' URLs (roughly, HTTPS or localhost)
+            if sfs not in ('same-origin', 'none'):
+                raise HTTPError(403, "Cross-origin request with unsafe method")
+
+        else:
+            # Fallback: The Origin or Referrer header gives the domain
+            # the request came from, Host should tell us where we're running.
+            src_origin = headers.get("Origin") or headers.get("Referrer")
+            if src_origin is None:
+                raise HTTPError(403, "No Origin/Referrer header with unsafe method")
+            src_scheme, src_netloc = urllib.parse.urlsplit(src_origin)[:2]
+            if src_scheme != self.request.protocol or src_netloc != self.request.host:
+                raise HTTPError(403, "Cross-origin request with unsafe method")
+
     def static_url(
         self, path: str, include_host: Optional[bool] = None, **kwargs: Any
     ) -> str:
@@ -1826,12 +1845,11 @@ async def _execute(
             }
             # If XSRF cookies are turned on, reject form submissions without
             # the proper cookie
-            if self.request.method not in (
-                "GET",
-                "HEAD",
-                "OPTIONS",
-            ) and self.application.settings.get("xsrf_cookies"):
-                self.check_xsrf_cookie()
+            if self.request.method not in ("GET", "HEAD", "OPTIONS"):
+                if self.application.settings.get("xsrf_cookies"):
+                    self.check_xsrf_cookie()
+                if self.application.settings.get("check_same_origin"):
+                    self.check_same_origin()
 
             result = self.prepare()
             if result is not None: