core.py 13 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386
  1. import logging
  2. import re
  3. from collections.abc import Iterable
  4. from datetime import timedelta
  5. from flask import current_app, request
  6. from werkzeug.datastructures import Headers, MultiDict
  7. LOG = logging.getLogger(__name__)
  8. # Response Headers
  9. ACL_ORIGIN = "Access-Control-Allow-Origin"
  10. ACL_METHODS = "Access-Control-Allow-Methods"
  11. ACL_ALLOW_HEADERS = "Access-Control-Allow-Headers"
  12. ACL_EXPOSE_HEADERS = "Access-Control-Expose-Headers"
  13. ACL_CREDENTIALS = "Access-Control-Allow-Credentials"
  14. ACL_MAX_AGE = "Access-Control-Max-Age"
  15. ACL_RESPONSE_PRIVATE_NETWORK = "Access-Control-Allow-Private-Network"
  16. # Request Header
  17. ACL_REQUEST_METHOD = "Access-Control-Request-Method"
  18. ACL_REQUEST_HEADERS = "Access-Control-Request-Headers"
  19. ACL_REQUEST_HEADER_PRIVATE_NETWORK = "Access-Control-Request-Private-Network"
  20. ALL_METHODS = ["GET", "HEAD", "POST", "OPTIONS", "PUT", "PATCH", "DELETE"]
  21. CONFIG_OPTIONS = [
  22. "CORS_ORIGINS",
  23. "CORS_METHODS",
  24. "CORS_ALLOW_HEADERS",
  25. "CORS_EXPOSE_HEADERS",
  26. "CORS_SUPPORTS_CREDENTIALS",
  27. "CORS_MAX_AGE",
  28. "CORS_SEND_WILDCARD",
  29. "CORS_AUTOMATIC_OPTIONS",
  30. "CORS_VARY_HEADER",
  31. "CORS_RESOURCES",
  32. "CORS_INTERCEPT_EXCEPTIONS",
  33. "CORS_ALWAYS_SEND",
  34. "CORS_ALLOW_PRIVATE_NETWORK",
  35. ]
  36. # Attribute added to request object by decorator to indicate that CORS
  37. # was evaluated, in case the decorator and extension are both applied
  38. # to a view.
  39. FLASK_CORS_EVALUATED = "_FLASK_CORS_EVALUATED"
  40. # Strange, but this gets the type of a compiled regex, which is otherwise not
  41. # exposed in a public API.
  42. RegexObject = type(re.compile(""))
  43. DEFAULT_OPTIONS = dict(
  44. origins="*",
  45. methods=ALL_METHODS,
  46. allow_headers="*",
  47. expose_headers=None,
  48. supports_credentials=False,
  49. max_age=None,
  50. send_wildcard=False,
  51. automatic_options=True,
  52. vary_header=True,
  53. resources=r"/*",
  54. intercept_exceptions=True,
  55. always_send=True,
  56. allow_private_network=False,
  57. )
  58. def parse_resources(resources):
  59. if isinstance(resources, dict):
  60. # To make the API more consistent with the decorator, allow a
  61. # resource of '*', which is not actually a valid regexp.
  62. resources = [(re_fix(k), v) for k, v in resources.items()]
  63. # Sort by regex length to provide consistency of matching and
  64. # to provide a proxy for specificity of match. E.G. longer
  65. # regular expressions are tried first.
  66. def pattern_length(pair):
  67. maybe_regex, _ = pair
  68. return len(get_regexp_pattern(maybe_regex))
  69. return sorted(resources, key=pattern_length, reverse=True)
  70. elif isinstance(resources, str):
  71. return [(re_fix(resources), {})]
  72. elif isinstance(resources, Iterable):
  73. return [(re_fix(r), {}) for r in resources]
  74. # Type of compiled regex is not part of the public API. Test for this
  75. # at runtime.
  76. elif isinstance(resources, RegexObject):
  77. return [(re_fix(resources), {})]
  78. else:
  79. raise ValueError("Unexpected value for resources argument.")
  80. def get_regexp_pattern(regexp):
  81. """
  82. Helper that returns regexp pattern from given value.
  83. :param regexp: regular expression to stringify
  84. :type regexp: _sre.SRE_Pattern or str
  85. :returns: string representation of given regexp pattern
  86. :rtype: str
  87. """
  88. try:
  89. return regexp.pattern
  90. except AttributeError:
  91. return str(regexp)
  92. def get_cors_origins(options, request_origin):
  93. origins = options.get("origins")
  94. wildcard = r".*" in origins
  95. # If the Origin header is not present terminate this set of steps.
  96. # The request is outside the scope of this specification.-- W3Spec
  97. if request_origin:
  98. LOG.debug("CORS request received with 'Origin' %s", request_origin)
  99. # If the allowed origins is an asterisk or 'wildcard', always match
  100. if wildcard and options.get("send_wildcard"):
  101. LOG.debug("Allowed origins are set to '*'. Sending wildcard CORS header.")
  102. return ["*"]
  103. # If the value of the Origin header is a case-sensitive match
  104. # for any of the values in list of origins
  105. elif try_match_any(request_origin, origins):
  106. LOG.debug(
  107. "The request's Origin header matches. Sending CORS headers.",
  108. )
  109. # Add a single Access-Control-Allow-Origin header, with either
  110. # the value of the Origin header or the string "*" as value.
  111. # -- W3Spec
  112. return [request_origin]
  113. else:
  114. LOG.debug("The request's Origin header does not match any of allowed origins.")
  115. return None
  116. elif options.get("always_send"):
  117. if wildcard:
  118. # If wildcard is in the origins, even if 'send_wildcard' is False,
  119. # simply send the wildcard. Unless supports_credentials is True,
  120. # since that is forbidden by the spec..
  121. # It is the most-likely to be correct thing to do (the only other
  122. # option is to return nothing, which almost certainly not what
  123. # the developer wants if the '*' origin was specified.
  124. if options.get("supports_credentials"):
  125. return None
  126. else:
  127. return ["*"]
  128. else:
  129. # Return all origins that are not regexes.
  130. return sorted([o for o in origins if not probably_regex(o)])
  131. # Terminate these steps, return the original request untouched.
  132. else:
  133. LOG.debug(
  134. "The request did not contain an 'Origin' header. This means the browser or client did not request CORS, ensure the Origin Header is set."
  135. )
  136. return None
  137. def get_allow_headers(options, acl_request_headers):
  138. if acl_request_headers:
  139. request_headers = [h.strip() for h in acl_request_headers.split(",")]
  140. # any header that matches in the allow_headers
  141. matching_headers = filter(lambda h: try_match_any(h, options.get("allow_headers")), request_headers)
  142. return ", ".join(sorted(matching_headers))
  143. return None
  144. def get_cors_headers(options, request_headers, request_method):
  145. origins_to_set = get_cors_origins(options, request_headers.get("Origin"))
  146. headers = MultiDict()
  147. if not origins_to_set: # CORS is not enabled for this route
  148. return headers
  149. for origin in origins_to_set:
  150. headers.add(ACL_ORIGIN, origin)
  151. headers[ACL_EXPOSE_HEADERS] = options.get("expose_headers")
  152. if options.get("supports_credentials"):
  153. headers[ACL_CREDENTIALS] = "true" # case sensitive
  154. if (
  155. ACL_REQUEST_HEADER_PRIVATE_NETWORK in request_headers
  156. and request_headers.get(ACL_REQUEST_HEADER_PRIVATE_NETWORK) == "true"
  157. ):
  158. allow_private_network = "true" if options.get("allow_private_network") else "false"
  159. headers[ACL_RESPONSE_PRIVATE_NETWORK] = allow_private_network
  160. # This is a preflight request
  161. # http://www.w3.org/TR/cors/#resource-preflight-requests
  162. if request_method == "OPTIONS":
  163. acl_request_method = request_headers.get(ACL_REQUEST_METHOD, "").upper()
  164. # If there is no Access-Control-Request-Method header or if parsing
  165. # failed, do not set any additional headers
  166. if acl_request_method and acl_request_method in options.get("methods"):
  167. # If method is not a case-sensitive match for any of the values in
  168. # list of methods do not set any additional headers and terminate
  169. # this set of steps.
  170. headers[ACL_ALLOW_HEADERS] = get_allow_headers(options, request_headers.get(ACL_REQUEST_HEADERS))
  171. headers[ACL_MAX_AGE] = options.get("max_age")
  172. headers[ACL_METHODS] = options.get("methods")
  173. else:
  174. LOG.info(
  175. "The request's Access-Control-Request-Method header does not match allowed methods. CORS headers will not be applied."
  176. )
  177. # http://www.w3.org/TR/cors/#resource-implementation
  178. if options.get("vary_header"):
  179. # Only set header if the origin returned will vary dynamically,
  180. # i.e. if we are not returning an asterisk, and there are multiple
  181. # origins that can be matched.
  182. if headers[ACL_ORIGIN] == "*":
  183. pass
  184. elif (
  185. len(options.get("origins")) > 1
  186. or len(origins_to_set) > 1
  187. or any(map(probably_regex, options.get("origins")))
  188. ):
  189. headers.add("Vary", "Origin")
  190. return MultiDict((k, v) for k, v in headers.items() if v)
  191. def set_cors_headers(resp, options):
  192. """
  193. Performs the actual evaluation of Flask-CORS options and actually
  194. modifies the response object.
  195. This function is used both in the decorator and the after_request
  196. callback
  197. """
  198. # If CORS has already been evaluated via the decorator, skip
  199. if hasattr(resp, FLASK_CORS_EVALUATED):
  200. LOG.debug("CORS have been already evaluated, skipping")
  201. return resp
  202. # Some libraries, like OAuthlib, set resp.headers to non Multidict
  203. # objects (Werkzeug Headers work as well). This is a problem because
  204. # headers allow repeated values.
  205. if not isinstance(resp.headers, Headers) and not isinstance(resp.headers, MultiDict):
  206. resp.headers = MultiDict(resp.headers)
  207. headers_to_set = get_cors_headers(options, request.headers, request.method)
  208. LOG.debug("Settings CORS headers: %s", str(headers_to_set))
  209. for k, v in headers_to_set.items():
  210. resp.headers.add(k, v)
  211. return resp
  212. def probably_regex(maybe_regex):
  213. if isinstance(maybe_regex, RegexObject):
  214. return True
  215. else:
  216. common_regex_chars = ["*", "\\", "]", "?", "$", "^", "[", "]", "(", ")"]
  217. # Use common characters used in regular expressions as a proxy
  218. # for if this string is in fact a regex.
  219. return any(c in maybe_regex for c in common_regex_chars)
  220. def re_fix(reg):
  221. """
  222. Replace the invalid regex r'*' with the valid, wildcard regex r'/.*' to
  223. enable the CORS app extension to have a more user friendly api.
  224. """
  225. return r".*" if reg == r"*" else reg
  226. def try_match_any(inst, patterns):
  227. return any(try_match(inst, pattern) for pattern in patterns)
  228. def try_match(request_origin, maybe_regex):
  229. """Safely attempts to match a pattern or string to a request origin."""
  230. if isinstance(maybe_regex, RegexObject):
  231. return re.match(maybe_regex, request_origin)
  232. elif probably_regex(maybe_regex):
  233. return re.match(maybe_regex, request_origin, flags=re.IGNORECASE)
  234. else:
  235. try:
  236. return request_origin.lower() == maybe_regex.lower()
  237. except AttributeError:
  238. return request_origin == maybe_regex
  239. def get_cors_options(appInstance, *dicts):
  240. """
  241. Compute CORS options for an application by combining the DEFAULT_OPTIONS,
  242. the app's configuration-specified options and any dictionaries passed. The
  243. last specified option wins.
  244. """
  245. options = DEFAULT_OPTIONS.copy()
  246. options.update(get_app_kwarg_dict(appInstance))
  247. if dicts:
  248. for d in dicts:
  249. options.update(d)
  250. return serialize_options(options)
  251. def get_app_kwarg_dict(appInstance=None):
  252. """Returns the dictionary of CORS specific app configurations."""
  253. app = appInstance or current_app
  254. # In order to support blueprints which do not have a config attribute
  255. app_config = getattr(app, "config", {})
  256. return {k.lower().replace("cors_", ""): app_config.get(k) for k in CONFIG_OPTIONS if app_config.get(k) is not None}
  257. def flexible_str(obj):
  258. """
  259. A more flexible str function which intelligently handles stringifying
  260. strings, lists and other iterables. The results are lexographically sorted
  261. to ensure generated responses are consistent when iterables such as Set
  262. are used.
  263. """
  264. if obj is None:
  265. return None
  266. elif not isinstance(obj, str) and isinstance(obj, Iterable):
  267. return ", ".join(str(item) for item in sorted(obj))
  268. else:
  269. return str(obj)
  270. def serialize_option(options_dict, key, upper=False):
  271. if key in options_dict:
  272. value = flexible_str(options_dict[key])
  273. options_dict[key] = value.upper() if upper else value
  274. def ensure_iterable(inst):
  275. """
  276. Wraps scalars or string types as a list, or returns the iterable instance.
  277. """
  278. if isinstance(inst, str) or not isinstance(inst, Iterable):
  279. return [inst]
  280. else:
  281. return inst
  282. def sanitize_regex_param(param):
  283. return [re_fix(x) for x in ensure_iterable(param)]
  284. def serialize_options(opts):
  285. """
  286. A helper method to serialize and processes the options dictionary.
  287. """
  288. options = (opts or {}).copy()
  289. for key in opts.keys():
  290. if key not in DEFAULT_OPTIONS:
  291. LOG.warning("Unknown option passed to Flask-CORS: %s", key)
  292. # Ensure origins is a list of allowed origins with at least one entry.
  293. options["origins"] = sanitize_regex_param(options.get("origins"))
  294. options["allow_headers"] = sanitize_regex_param(options.get("allow_headers"))
  295. # This is expressly forbidden by the spec. Raise a value error so people
  296. # don't get burned in production.
  297. if r".*" in options["origins"] and options["supports_credentials"] and options["send_wildcard"]:
  298. raise ValueError(
  299. "Cannot use supports_credentials in conjunction with"
  300. "an origin string of '*'. See: "
  301. "http://www.w3.org/TR/cors/#resource-requests"
  302. )
  303. serialize_option(options, "expose_headers")
  304. serialize_option(options, "methods", upper=True)
  305. if isinstance(options.get("max_age"), timedelta):
  306. options["max_age"] = str(int(options["max_age"].total_seconds()))
  307. return options