Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
41 changes: 23 additions & 18 deletions flask_cors/core.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
:copyright: (c) 2016 by Cory Dolphin.
:license: MIT, see LICENSE for more details.
"""
from difflib import Match
import re
import logging
try:
Expand All @@ -18,7 +19,11 @@
from datetime import timedelta
from six import string_types
from flask import request, current_app
from werkzeug.datastructures import Headers, MultiDict
from werkzeug.datastructures import EnvironHeaders, Headers, MultiDict
from flask.app import Flask
from flask.wrappers import Response
from typing import Any, Dict, List, Optional, Pattern, Set, Union
from werkzeug.local import LocalProxy

LOG = logging.getLogger(__name__)

Expand Down Expand Up @@ -63,7 +68,7 @@
always_send=True)


def parse_resources(resources):
def parse_resources(resources: Any) -> List[Any]:
if isinstance(resources, dict):
# To make the API more consistent with the decorator, allow a
# resource of '*', which is not actually a valid regexp.
Expand Down Expand Up @@ -95,7 +100,7 @@ def pattern_length(pair):
raise ValueError("Unexpected value for resources argument.")


def get_regexp_pattern(regexp):
def get_regexp_pattern(regexp: Union[Pattern, str]) -> str:
"""
Helper that returns regexp pattern from given value.

Expand All @@ -110,7 +115,7 @@ def get_regexp_pattern(regexp):
return str(regexp)


def get_cors_origins(options, request_origin):
def get_cors_origins(options: Dict[str, Any], request_origin: Optional[str]) -> Optional[List[str]]:
origins = options.get('origins')
wildcard = r'.*' in origins

Expand Down Expand Up @@ -158,7 +163,7 @@ def get_cors_origins(options, request_origin):
return None


def get_allow_headers(options, acl_request_headers):
def get_allow_headers(options: Dict[str, Optional[Union[str, List[str], bool, int]]], acl_request_headers: Optional[str]) -> Optional[str]:
if acl_request_headers:
request_headers = [h.strip() for h in acl_request_headers.split(',')]

Expand All @@ -173,7 +178,7 @@ def get_allow_headers(options, acl_request_headers):
return None


def get_cors_headers(options, request_headers, request_method):
def get_cors_headers(options: Dict[str, Any], request_headers: EnvironHeaders, request_method: str) -> MultiDict:
origins_to_set = get_cors_origins(options, request_headers.get('Origin'))
headers = MultiDict()

Expand Down Expand Up @@ -221,7 +226,7 @@ def get_cors_headers(options, request_headers, request_method):
return MultiDict((k, v) for k, v in headers.items() if v)


def set_cors_headers(resp, options):
def set_cors_headers(resp: Response, options: Dict[str, Any]) -> Response:
"""
Performs the actual evaluation of Flask-CORS options and actually
modifies the response object.
Expand Down Expand Up @@ -251,7 +256,7 @@ def set_cors_headers(resp, options):

return resp

def probably_regex(maybe_regex):
def probably_regex(maybe_regex: Union[Pattern, str]) -> bool:
if isinstance(maybe_regex, RegexObject):
return True
else:
Expand All @@ -260,19 +265,19 @@ def probably_regex(maybe_regex):
# for if this string is in fact a regex.
return any((c in maybe_regex for c in common_regex_chars))

def re_fix(reg):
def re_fix(reg: Union[Pattern, str]) -> Union[Pattern, str]:
"""
Replace the invalid regex r'*' with the valid, wildcard regex r'/.*' to
enable the CORS app extension to have a more user friendly api.
"""
return r'.*' if reg == r'*' else reg


def try_match_any(inst, patterns):
def try_match_any(inst: str, patterns: List[Union[str, Pattern]]) -> bool:
return any(try_match(inst, pattern) for pattern in patterns)


def try_match(request_origin, maybe_regex):
def try_match(request_origin: str, maybe_regex: Union[Pattern, str]) -> Optional[Union[Match, bool]]:
"""Safely attempts to match a pattern or string to a request origin."""
if isinstance(maybe_regex, RegexObject):
return re.match(maybe_regex, request_origin)
Expand All @@ -285,7 +290,7 @@ def try_match(request_origin, maybe_regex):
return request_origin == maybe_regex


def get_cors_options(appInstance, *dicts):
def get_cors_options(appInstance: Union[LocalProxy, Flask], *dicts) -> Dict[str, Any]:
"""
Compute CORS options for an application by combining the DEFAULT_OPTIONS,
the app's configuration-specified options and any dictionaries passed. The
Expand All @@ -300,7 +305,7 @@ def get_cors_options(appInstance, *dicts):
return serialize_options(options)


def get_app_kwarg_dict(appInstance=None):
def get_app_kwarg_dict(appInstance: Optional[Union[LocalProxy, Flask]]=None) -> Dict[Any, Any]:
"""Returns the dictionary of CORS specific app configurations."""
app = (appInstance or current_app)

Expand All @@ -314,7 +319,7 @@ def get_app_kwarg_dict(appInstance=None):
}


def flexible_str(obj):
def flexible_str(obj: Optional[Union[List[str], str]]) -> Optional[str]:
"""
A more flexible str function which intelligently handles stringifying
strings, lists and other iterables. The results are lexographically sorted
Expand All @@ -330,13 +335,13 @@ def flexible_str(obj):
return str(obj)


def serialize_option(options_dict, key, upper=False):
def serialize_option(options_dict: Dict[str, Any], key: str, upper: bool=False) -> None:
if key in options_dict:
value = flexible_str(options_dict[key])
options_dict[key] = value.upper() if upper else value


def ensure_iterable(inst):
def ensure_iterable(inst: Union[List[str], Pattern, str, Set[str]]) -> Union[List[str], List[Pattern], Set[str]]:
"""
Wraps scalars or string types as a list, or returns the iterable instance.
"""
Expand All @@ -347,11 +352,11 @@ def ensure_iterable(inst):
else:
return inst

def sanitize_regex_param(param):
def sanitize_regex_param(param: Union[List[str], Set[str], Pattern, str]) -> List[Union[str, Pattern]]:
return [re_fix(x) for x in ensure_iterable(param)]


def serialize_options(opts):
def serialize_options(opts: Dict[str, Any]) -> Dict[str, Any]:
"""
A helper method to serialize and processes the options dictionary.
"""
Expand Down
3 changes: 2 additions & 1 deletion flask_cors/decorator.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,10 +12,11 @@
from functools import update_wrapper
from flask import make_response, request, current_app
from .core import *
from typing import Callable

LOG = logging.getLogger(__name__)

def cross_origin(*args, **kwargs):
def cross_origin(*args, **kwargs) -> Callable:
"""
This function is the decorator which is used to wrap a Flask route with.
In the simplest case, simply use the default parameters to allow all
Expand Down
9 changes: 6 additions & 3 deletions flask_cors/extension.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,9 @@
"""
from flask import request
from .core import *
from flask.app import Flask
from typing import Any, Callable, List, Optional

try:
from urllib.parse import unquote_plus
except ImportError:
Expand Down Expand Up @@ -131,12 +134,12 @@ class CORS(object):
:type vary_header: bool
"""

def __init__(self, app=None, **kwargs):
def __init__(self, app: Optional[Flask]=None, **kwargs) -> None:
self._options = kwargs
if app is not None:
self.init_app(app, **kwargs)

def init_app(self, app, **kwargs):
def init_app(self, app: Flask, **kwargs) -> None:
# The resources and options may be specified in the App Config, the CORS constructor
# or the kwargs to the call to init_app.
options = get_cors_options(app, self._options, kwargs)
Expand Down Expand Up @@ -175,7 +178,7 @@ def wrapped_function(*args, **kwargs):
app.handle_user_exception = _after_request_decorator(
app.handle_user_exception)

def make_after_request_function(resources):
def make_after_request_function(resources: List[Any]) -> Callable:
def cors_after_request(resp):
# If CORS headers are set in a view decorator, pass
if resp.headers is not None and resp.headers.get(ACL_ORIGIN):
Expand Down
4 changes: 2 additions & 2 deletions tests/base_test.py
Original file line number Diff line number Diff line change
Expand Up @@ -16,7 +16,7 @@


class FlaskCorsTestCase(unittest.TestCase):
def shortDescription(self):
def shortDescription(self) -> str:
"""
Get's the one liner description to be displayed.
Source:
Expand All @@ -37,7 +37,7 @@ def iter_responses(self, path, verbs=['get', 'head', 'options'], **kwargs):
for verb in verbs:
yield self._request(verb.lower(), path, **kwargs)

def _request(self, verb, *args, **kwargs):
def _request(self, verb: str, *args, **kwargs):
_origin = kwargs.pop('origin', None)
headers = kwargs.pop('headers', {})
if _origin:
Expand Down
4 changes: 2 additions & 2 deletions tests/core/test_override_headers.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from flask_cors.core import *

class ResponseHeadersOverrideTestCaseIntegration(FlaskCorsTestCase):
def setUp(self):
def setUp(self) -> None:
self.app = Flask(__name__)
CORS(self.app)

Expand All @@ -22,7 +22,7 @@ def index():
response = Response(headers={"custom": "dictionary"})
return 'Welcome'

def test_override_headers(self):
def test_override_headers(self) -> None:
'''
Ensure we work even if response.headers is set to something other than a MultiDict.
'''
Expand Down
14 changes: 7 additions & 7 deletions tests/decorator/test_allow_headers.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,7 @@
from flask_cors.core import *

class AllowHeadersTestCaseIntegration(FlaskCorsTestCase):
def setUp(self):
def setUp(self) -> None:
self.app = Flask(__name__)

@self.app.route('/test_default')
Expand All @@ -32,19 +32,19 @@ def test_allow_headers():
def test_allow_headers_regex():
return 'Welcome!'

def test_default(self):
def test_default(self) -> None:
for resp in self.iter_responses('/test_default'):
self.assertTrue(resp.headers.get(ACL_ALLOW_HEADERS) is None,
"Default should have no allowed headers")

def test_allow_headers_no_request_headers(self):
def test_allow_headers_no_request_headers(self) -> None:
'''
No ACL_REQUEST_HEADERS sent, ACL_ALLOW_HEADERS should be empty
'''
resp = self.preflight('/test_allow_headers', origin='www.example.com')
self.assertEqual(resp.headers.get(ACL_ALLOW_HEADERS), None)

def test_allow_headers_with_request_headers(self):
def test_allow_headers_with_request_headers(self) -> None:
'''
If there is an Access-Control-Request-Method header in the request
and Access-Control-Request-Method is allowed for cross origin
Expand All @@ -58,7 +58,7 @@ def test_allow_headers_with_request_headers(self):
self.assertEqual(resp.headers.get(ACL_ALLOW_HEADERS),
'X-Example-Header-A')

def test_allow_headers_with_request_headers_case_insensitive(self):
def test_allow_headers_with_request_headers_case_insensitive(self) -> None:
'''
HTTP headers are case insensitive. We should respect that
and match regardless of case, returning the casing sent by
Expand All @@ -70,7 +70,7 @@ def test_allow_headers_with_request_headers_case_insensitive(self):
self.assertEqual(resp.headers.get(ACL_ALLOW_HEADERS),
'X-Example-header-a')

def test_allow_headers_with_unmatched_request_headers(self):
def test_allow_headers_with_unmatched_request_headers(self) -> None:
'''
If every element in the Access-Control-Request-Headers is not an
allowed header, then the matching headers should be returned.
Expand All @@ -87,7 +87,7 @@ def test_allow_headers_with_unmatched_request_headers(self):
self.assertEqual(resp.headers.get(ACL_ALLOW_HEADERS),
'X-Example-Header-A')

def test_allow_headers_regex(self):
def test_allow_headers_regex(self) -> None:
'''
If every element in the Access-Control-Request-Headers is not an
allowed header, then the matching headers should be returned.
Expand Down
8 changes: 4 additions & 4 deletions tests/decorator/test_credentials.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,7 +17,7 @@


class SupportsCredentialsCase(FlaskCorsTestCase):
def setUp(self):
def setUp(self) -> None:
self.app = Flask(__name__)

@self.app.route('/test_credentials_supported')
Expand All @@ -35,14 +35,14 @@ def test_credentials_unsupported():
def test_default():
return 'Open!'

def test_credentials_supported(self):
def test_credentials_supported(self) -> None:
''' The specified route should return the
Access-Control-Allow-Credentials header.
'''
resp = self.get('/test_credentials_supported', origin='www.example.com')
self.assertEqual(resp.headers.get(ACL_CREDENTIALS), 'true')

def test_default(self):
def test_default(self) -> None:
''' The default behavior should be to disallow credentials.
'''
resp = self.get('/test_default', origin='www.example.com')
Expand All @@ -51,7 +51,7 @@ def test_default(self):
resp = self.get('/test_default')
self.assertFalse(ACL_CREDENTIALS in resp.headers)

def test_credentials_unsupported(self):
def test_credentials_unsupported(self) -> None:
''' The default behavior should be to disallow credentials.
'''
resp = self.get('/test_credentials_unsupported', origin='www.example.com')
Expand Down
4 changes: 2 additions & 2 deletions tests/decorator/test_duplicate_headers.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@


class AllowsMultipleHeaderEntries(FlaskCorsTestCase):
def setUp(self):
def setUp(self) -> None:
self.app = Flask(__name__)

@self.app.route('/test_multiple_set_cookie_headers')
Expand All @@ -25,7 +25,7 @@ def test_multiple_set_cookie_headers():
resp.headers.add('set-cookie', 'bar')
return resp

def test_multiple_set_cookie_headers(self):
def test_multiple_set_cookie_headers(self) -> None:
resp = self.get('/test_multiple_set_cookie_headers')
self.assertEqual(len(resp.headers.getlist('set-cookie')), 2)

Expand Down
Loading