Skip to content

Commit

Permalink
Merge pull request #39 from geoadmin/develop
Browse files Browse the repository at this point in the history
Release v2.1.0
  • Loading branch information
ltshb authored Nov 15, 2021
2 parents 8ad005d + c7c1446 commit 3765538
Show file tree
Hide file tree
Showing 7 changed files with 83 additions and 15 deletions.
2 changes: 2 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -180,3 +180,5 @@ The service is configured by Environment Variable:
|-------------|-----------------------|----------------------------------------|
| LOGGING_CFG | `logging-cfg-local.yml` | Logging configuration file |
| ALLOWED_DOMAINS | `.*` | Comma separated list of regex that are allowed as domain in Origin header |
| CACHE_CONTROL | `public, max-age=31536000` | Cache Control header value of the GET /generate endpoint |
| CACHE_CONTROL_4XX | `public, max-age=3600` | Cache Control header for 4XX responses |
23 changes: 21 additions & 2 deletions app/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,10 +8,13 @@
from flask import abort
from flask import g
from flask import request
from flask.helpers import url_for

from app import settings
from app.helpers.utils import ALLOWED_DOMAINS_PATTERN
from app.helpers.utils import make_error_msg
from app.settings import ALLOWED_DOMAINS_PATTERN
from app.settings import CACHE_CONTROL
from app.settings import CACHE_CONTROL_4XX

logger = logging.getLogger(__name__)
route_logger = logging.getLogger('app.routes')
Expand Down Expand Up @@ -44,12 +47,28 @@ def validate_origin():
# Add CORS Headers to all request
@app.after_request
def add_cors_header(response):
# Do not add CORS header to internal /checker endpoint.
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.headers['Origin']
response.headers['Access-Control-Allow-Methods'] = 'GET, POST, OPTIONS'
response.headers['Access-Control-Allow-Methods'] = 'GET, HEAD, OPTIONS'
response.headers['Access-Control-Allow-Headers'] = '*'
return response


@app.after_request
def add_cache_control_header(response):
# For /checker route we let the frontend proxy decide how to cache it.
if request.method == 'GET' and request.endpoint != 'checker':
if response.status_code >= 400:
response.headers.set('Cache-Control', CACHE_CONTROL_4XX)
else:
response.headers.set('Cache-Control', CACHE_CONTROL)
return response


Expand Down
2 changes: 1 addition & 1 deletion app/helpers/url.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,8 +3,8 @@

from flask import abort

from app.helpers.utils import ALLOWED_DOMAINS_PATTERN
from app.helpers.utils import logger
from app.settings import ALLOWED_DOMAINS_PATTERN


def validate_url(url):
Expand Down
4 changes: 0 additions & 4 deletions app/helpers/utils.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,12 +7,8 @@
from flask import jsonify
from flask import make_response

from app.settings import ALLOWED_DOMAINS

logger = logging.getLogger(__name__)

ALLOWED_DOMAINS_PATTERN = f"({'|'.join(ALLOWED_DOMAINS)})"


def make_error_msg(code, msg):
return make_response(jsonify({'success': False, 'error': {'code': code, 'message': msg}}), code)
Expand Down
2 changes: 1 addition & 1 deletion app/routes.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@


@app.route('/checker', methods=['GET'])
def check():
def checker():
return make_response(jsonify({'success': True, 'message': 'OK', 'version': APP_VERSION}))


Expand Down
4 changes: 4 additions & 0 deletions app/settings.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,4 +18,8 @@
raise RuntimeError("Environment variable $ALLOWED_DOMAINS was not set")

ALLOWED_DOMAINS = ALLOWED_DOMAINS_STRING.split(',')
ALLOWED_DOMAINS_PATTERN = f"({'|'.join(ALLOWED_DOMAINS)})"
TRAP_HTTP_EXCEPTIONS = True

CACHE_CONTROL = os.getenv('CACHE_CONTROL', 'public, max-age=31536000')
CACHE_CONTROL_4XX = os.getenv('CACHE_CONTROL_4XX', 'public, max-age=3600')
61 changes: 54 additions & 7 deletions tests/unit_tests/test_qrcode.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,5 @@
import io
import re
import unittest
from urllib.parse import quote

Expand All @@ -8,6 +9,7 @@
from flask import url_for

from app import app
from app.settings import ALLOWED_DOMAINS_PATTERN
from app.version import APP_VERSION


Expand All @@ -23,15 +25,40 @@ def setUp(self):
'Origin': 'some_random_domain'
}

def assertCors(self, response, 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'])
)
self.assertIn('Access-Control-Allow-Methods', response.headers)
self.assertListEqual(
sorted(['GET', 'HEAD', 'OPTIONS']),
sorted(
map(
lambda m: m.strip(),
response.headers['Access-Control-Allow-Methods'].split(',')
)
)
)
self.assertIn('Access-Control-Allow-Headers', response.headers)
self.assertEqual(response.headers['Access-Control-Allow-Headers'], '*')

def test_checker(self):
response = self.app.get(url_for('check'), headers=self.valid_origin_header)
response = self.app.get(url_for('checker'), headers=self.valid_origin_header)
self.assertEqual(response.status_code, 200)
self.assertNotIn('Cache-Control', response.headers)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(response.json, {"message": "OK", "success": True, "version": APP_VERSION})

def test_generate_errors(self):
response = self.app.get(url_for('generate_get'))
self.assertEqual(response.status_code, 403, msg="ORIGIN must be set")
self.assertCors(response, check_origin=False)
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(
response.json, {
Expand All @@ -42,6 +69,7 @@ def test_generate_errors(self):
)
response = self.app.post(url_for('generate_get'), headers=self.valid_origin_header)
self.assertEqual(response.status_code, 405, msg="POST method is not allowed")
self.assertCors(response)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(
response.json,
Expand All @@ -58,6 +86,11 @@ def test_generate_errors(self):
self.assertEqual(
response.status_code, 400, msg="Should respond with a 400 when URL param is missing"
)
self.assertCors(response)
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(
response.json, {
Expand All @@ -74,18 +107,24 @@ def test_generate_domain_restriction(self):
headers=self.valid_origin_header
)
self.assertEqual(response.status_code, 200)
self.assertCors(response)
self.assertEqual(response.content_type, "image/png")
self.assertEqual(response.headers['Access-Control-Allow-Origin'], "some_random_domain")
self.assertEqual(response.headers['Access-Control-Allow-Methods'], "GET, POST, OPTIONS")
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)

response = self.app.get(
url_for('generate_get'),
query_string={'url': 'https://www.example.com/test'},
headers=self.valid_origin_header
)
self.assertEqual(response.status_code, 400, msg="Domain restriction not applied")
self.assertEqual(response.headers['Access-Control-Allow-Origin'], "some_random_domain")
self.assertEqual(response.headers['Access-Control-Allow-Methods'], "GET, POST, OPTIONS")
self.assertCors(response)
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=3600', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(
response.json, {
Expand All @@ -101,6 +140,11 @@ def test_generate_domain_restriction(self):
headers={"Origin": "www.example.com"}
)
self.assertEqual(response.status_code, 403, msg="Domain restriction not applied")
self.assertCors(response, check_origin=False)
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)
self.assertEqual(response.content_type, "application/json")
self.assertEqual(
response.json, {
Expand All @@ -119,9 +163,12 @@ def test_generate(self):
headers=self.valid_origin_header
)
self.assertEqual(response.status_code, 200)
self.assertCors(response)
self.assertEqual(response.content_type, "image/png")
self.assertEqual(response.headers['Access-Control-Allow-Origin'], "some_random_domain")
self.assertEqual(response.headers['Access-Control-Allow-Methods'], "GET, POST, OPTIONS")
self.assertIn('Cache-Control', response.headers, msg="Cache control header missing")
self.assertIn(
'max-age=', response.headers['Cache-Control'], msg="Cache Control max-age not set"
)

# decode the qrcode image into the url
image = io.BytesIO(response.data)
Expand Down

0 comments on commit 3765538

Please sign in to comment.