Merge pull request #57 from geoadmin/develop

New Release v1.5.1 - #patch

.env.testing

@@ -6,4 +6,4 @@ AWS_S3_REGION_NAME=wonderland
 # Fake regions will cause a crash when mocking.
 AWS_DB_REGION_NAME=us-east-1
 AWS_DB_TABLE_NAME=test-db
-ALLOWED_DOMAINS=.*\.geo\.admin\.ch
+ALLOWED_DOMAINS=.*\.geo\.admin\.ch,http://localhost

app/__init__.py

@@ -24,6 +24,10 @@
 app = Flask(__name__)
 
 
+def is_domain_allowed(domain):
+    return re.match(ALLOWED_DOMAINS_PATTERN, domain) is not None
+
+
 # Add quick log of the routes used to all request.
 # Important: this should be the first before_request method, to ensure
 # a failure in another pre request method would stop logging.
@@ -40,11 +44,10 @@ def add_cors_header(response):
     if request.endpoint == 'checker':
         return response
 
-    if (
-        'Origin' in request.headers and
-        re.match(ALLOWED_DOMAINS_PATTERN, request.headers['Origin'])
-    ):
+    response.headers['Access-Control-Allow-Origin'] = request.host_url
+    if 'Origin' in request.headers and is_domain_allowed(request.headers['Origin']):
         response.headers['Access-Control-Allow-Origin'] = request.headers['Origin']
+    response.headers['Vary'] = 'Origin'
 
     # Always add the allowed methods.
     response.headers.set(
@@ -70,11 +73,26 @@ def add_cache_control_header(response):
 # Reject request from non allowed origins
 @app.before_request
 def validate_origin():
-    if 'Origin' not in request.headers:
-        logger.error('Origin header is not set')
+    # The Origin headers is automatically set by the browser and cannot be changed by the javascript
+    # application. Unfortunately this header is only set if the request comes from another origin.
+    # Sec-Fetch-Site header is set to `same-origin` by most of the browser except by Safari !
+    # The best protection would be to use the Sec-Fetch-Site and Origin header, however this is
+    # not supported by Safari. Therefore we added a fallback to the Referer header for Safari.
+    sec_fetch_site = request.headers.get('Sec-Fetch-Site', None)
+    origin = request.headers.get('Origin', None)
+    referrer = request.headers.get('Referer', None)
+
+    if origin is None and referrer is None and sec_fetch_site is None:
+        logger.error('Referer and/or Origin and/or Sec-Fetch-Site headers not set')
+        abort(403, 'Permission denied')
+    if origin is not None and not is_domain_allowed(origin):
+        logger.error('Origin=%s is not allowed', origin)
+        abort(403, 'Permission denied')
+    if referrer is not None and not is_domain_allowed(referrer):
+        logger.error('Referer=%s is not allowed', referrer)
         abort(403, 'Permission denied')
-    if not re.match(ALLOWED_DOMAINS_PATTERN, request.headers['Origin']):
-        logger.error('Origin=%s is not allowed', request.headers['Origin'])
+    if sec_fetch_site is not None and sec_fetch_site != 'same-origin':
+        logger.error('Sec-Fetch-Site=%s is not allowed', sec_fetch_site)
         abort(403, 'Permission denied')
 
 

tests/unit_tests/base.py

@@ -139,12 +139,13 @@ def tearDown(self):
         for kml in self.kmls:
             self.delete_test_kml(kml['id'], kml['admin_id'])
 
-    def assertCors(self, response, expected_allowed_methods, check_origin=True):  # pylint: disable=invalid-name
-        if check_origin:
-            self.assertIn('Access-Control-Allow-Origin', response.headers)
-            self.assertTrue(
-                re.match(ALLOWED_DOMAINS_PATTERN, response.headers['Access-Control-Allow-Origin'])
-            )
+    def assertCors(self, response, expected_allowed_methods):  # pylint: disable=invalid-name
+        self.assertIn('Access-Control-Allow-Origin', response.headers)
+        self.assertIsNotNone(
+            re.match(ALLOWED_DOMAINS_PATTERN, response.headers['Access-Control-Allow-Origin']),
+            msg=f"Access-Control-Allow-Origin={response.headers['Access-Control-Allow-Origin']} "
+            f"doesn't match {ALLOWED_DOMAINS_PATTERN}"
+        )
         self.assertIn('Access-Control-Allow-Methods', response.headers)
         self.assertListEqual(
             sorted(expected_allowed_methods),

tests/unit_tests/test_routes.py

@@ -95,7 +95,7 @@ def test_valid_kml_post_non_allowed_origin(self):
             headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['GET', 'HEAD', 'POST', 'OPTIONS'], check_origin=False)
+        self.assertCors(response, ['GET', 'HEAD', 'POST', 'OPTIONS'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -223,7 +223,7 @@ def test_get_metadata_non_allowed_origin(self):
             url_for('get_kml_metadata', kml_id=id_to_fetch), headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertIn('Cache-Control', response.headers)
         self.assertIn('max-age=3600', response.headers['Cache-Control'])
         self.assertNotIn('Expire', response.headers)
@@ -354,7 +354,7 @@ def test_valid_kml_put_non_allowed_origin(self):
             headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -367,7 +367,7 @@ def test_valid_kml_put_non_allowed_admin_id(self):
             headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -380,7 +380,7 @@ def test_valid_kml_put_missing_admin_id(self):
             headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -442,7 +442,7 @@ def test_kml_delete_non_allowed_origin(self):
             headers=self.origin_headers["bad"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -456,7 +456,7 @@ def test_kml_delete_non_allowed_admin_id(self):
             headers=self.origin_headers["allowed"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")
 
@@ -469,7 +469,7 @@ def test_kml_delete_missing_admin_id(self):
             headers=self.origin_headers["allowed"]
         )
         self.assertEqual(response.status_code, 403)
-        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'], check_origin=False)
+        self.assertCors(response, ['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PUT'])
         self.assertEqual(response.content_type, "application/json")
         self.assertEqual(response.json["error"]["message"], "Permission denied")