Coverage for /home/runner/work/viur-core/viur-core/viur/src/viur/core/request.py: 6%

1""" 

2 This module implements the WSGI (Web Server Gateway Interface) layer for ViUR. This is the main entry 

3 point for incomming http requests. The main class is the :class:BrowserHandler. Each request will get it's 

4 own instance of that class which then holds the reference to the request and response object. 

5 Additionally, this module defines the RequestValidator interface which provides a very early hook into the 

6 request processing (useful for global ratelimiting, DDoS prevention or access control). 

7""" 

8import datetime 

9import fnmatch 

10import json 

11import logging 

12import os 

13import re 

14import time 

15import traceback 

16import typing as t 

17import unicodedata 

18from abc import ABC, abstractmethod 

19from urllib import parse 

20from urllib.parse import quote, unquote, urljoin, urlparse 

21 

22import webob 

23 

24from viur.core import current, db, errors, session, utils 

25from viur.core.config import conf 

26from viur.core.logging import client as loggingClient, requestLogger, requestLoggingRessource 

27from viur.core.module import Method 

28from viur.core.securityheaders import _build_reporting_endpoints_header, extendCsp 

29from viur.core.tasks import _appengineServiceIPs 

30 

31TEMPLATE_STYLE_KEY = "style" 

32 

33 

34class RequestValidator(ABC): 

35 """ 

36 RequestValidators can be used to validate a request very early on. If the validate method returns a tuple, 

37 the request is aborted. Can be used to block requests from bots. 

38 

39 To register or remove a validator, access it in main.py through 

40 :attr: viur.core.request.Router.requestValidators 

41 """ 

42 # Internal name to trace which validator aborted the request 

43 name = "RequestValidator" 

44 

45 @staticmethod 

46 @abstractmethod 

47 def validate(request: 'BrowseHandler') -> t.Optional[tuple[int, str, str]]: 

48 """ 

49 The function that checks the current request. If the request is valid, simply return None. 

50 If the request should be blocked, it must return a tuple of 

51 - The HTTP status code (as int) 

52 - The Description of that status code (eg "Forbidden") 

53 - The Response Body (can be a simple string or an HTML-Page) 

54 :param request: The Request instance to check 

55 :return: None on success, an Error-Tuple otherwise 

56 """ 

57 raise NotImplementedError() 

58 

59 

60class FetchMetaDataValidator(RequestValidator): 

61 """ 

62 This validator examines the headers "Sec-Fetch-Site", "sec-fetch-mode" and "sec-fetch-dest" as 

63 recommended by https://web.dev/fetch-metadata/ 

64 """ 

65 name = "FetchMetaDataValidator" 

66 

67 @staticmethod 

68 def validate(request: 'BrowseHandler') -> t.Optional[tuple[int, str, str]]: 

69 """ 

70 This validator examines the headers "sec-fetch-site", 

71 "sec-fetch-mode" and "sec-fetch-dest" as recommended 

72 by https://web.dev/fetch-metadata/ 

73 """ 

74 headers = request.request.headers 

75 

76 match headers.get("sec-fetch-site"): 

77 case None | "same-origin" | "same-site" | "none": 

78 # Browser didn't send "sec-fetch-site", or the request is 

79 # same-origin, same-site (e.g. subdomain or redirect within the 

80 # same registrable domain) or browser-initiated ("none"). 

81 # These are allowed per the reference policy on 

82 # https://web.dev/fetch-metadata/ 

83 return None 

84 case _: 

85 # Incoming cross-site request: allow only simple top-level 

86 # navigation GET requests, except <object> and <embed>. 

87 if ( 

88 not request.isPostRequest 

89 and headers.get("sec-fetch-mode") == "navigate" 

90 and headers.get('sec-fetch-dest') not in ("object", "embed") 

91 ): 

92 return None 

93 

94 return 403, "Forbidden", "Request rejected due to fetch metadata" 

95 

96 

97class Router: 

98 """ 

99 This class accepts the requests, collect its parameters and routes the request 

100 to its destination function. 

101 The basic control flow is 

102 - Setting up internal variables 

103 - Running the Request validators 

104 - Emitting the headers (especially the security related ones) 

105 - Run the TLS check (ensure it's a secure connection or check if the URL is whitelisted) 

106 - Load or initialize a new session 

107 - Set up i18n (choosing the language etc) 

108 - Run the request preprocessor (if any) 

109 - Run before_request hooks (if any) 

110 - Normalize & sanity check the parameters 

111 - Resolve the exposed function and call it 

112 - Save the session / tear down the request 

113 - Run after_request hooks (if any) 

114 - Return the response generated 

115 

116 

117 :warning: Don't instantiate! Don't subclass! DON'T TOUCH! ;) 

118 """ 

119 

120 # List of requestValidators used to preflight-check an request before it's being dispatched within ViUR 

121 requestValidators = [FetchMetaDataValidator] 

122 

123 before_request_funcs: t.ClassVar[list[t.Callable[[], None]]] = [] 

124 after_request_funcs: t.ClassVar[list[t.Callable[[], None]]] = [] 

125 

126 def __init__(self, environ: dict): 

127 super().__init__() 

128 self.startTime = time.time() 

129 

130 self.request = webob.Request(environ) 

131 self.response = webob.Response() 

132 

133 self.maxLogLevel = logging.DEBUG 

134 self._traceID = \ 

135 self.request.headers.get("X-Cloud-Trace-Context", "").split("/")[0] or utils.string.random() 

136 self.is_deferred = False 

137 self.path = "" 

138 self.path_list = () 

139 

140 self.skey_checked = False # indicates whether @skey-decorator-check has already performed within a request 

141 self.internalRequest = False 

142 self.disableCache = False # Shall this request bypass the caches? 

143 self.pendingTasks = [] 

144 self.args = () 

145 self.kwargs = {} 

146 self.context = {} 

147 self.template_style: str | None = None 

148 self.cors_headers = () 

149 

150 # Check if it's a HTTP-Method we support 

151 self.method = self.request.method.lower() 

152 self.isPostRequest = self.method == "post" 

153 self.isSSLConnection = self.request.host_url.lower().startswith("https://") # We have an encrypted channel 

154 

155 db.current_db_access_log.set(set()) 

156 

157 # Set context variables 

158 current.language.set(conf.i18n.default_language) 

159 current.request.set(self) 

160 current.session.set(session.Session()) 

161 current.request_data.set({}) 

162 

163 for fn in Router.before_request_funcs: 

164 fn() 

165 

166 # Process actual request 

167 self._process() 

168 

169 for fn in Router.after_request_funcs: 

170 fn() 

171 

172 self._cors() 

173 

174 # Unset context variables 

175 current.language.set(None) 

176 current.request_data.set(None) 

177 current.session.set(None) 

178 current.request.set(None) 

179 current.user.set(None) 

180 

181 @property 

182 def isDevServer(self) -> bool: 

183 import warnings 

184 msg = "Use of `isDevServer` is deprecated; Use `conf.instance.is_dev_server` instead!" 

185 warnings.warn(msg, DeprecationWarning, stacklevel=2) 

186 logging.warning(msg) 

187 return conf.instance.is_dev_server 

188 

189 def _select_language(self, path: str) -> str: 

190 """ 

191 Tries to select the best language for the current request. Depending on the value of 

192 conf.i18n.language_method, we'll either try to load it from the session, determine it by the domain 

193 or extract it from the URL. 

194 """ 

195 

196 def get_language_from_header() -> str | None: 

197 if not (accept_language := self.request.headers.get("accept-language")): 

198 return None 

199 languages = accept_language.split(",") 

200 locale_q_pairs = [] 

201 

202 for language in languages: 

203 if language.split(";")[0] == language: 

204 # no q => q = 1 

205 locale_q_pairs.append((language.strip(), "1")) 

206 else: 

207 try: 

208 locale = language.split(";")[0].strip() 

209 q = language.split(";")[1].split("=")[1] 

210 locale_q_pairs.append((locale, q)) 

211 except IndexError: 

212 continue # skip language 

213 locale_q_pairs.sort(key=lambda pair: pair[1], reverse=True) # sort by Quality values 

214 for locale_q_pair in locale_q_pairs: 

215 if "-" in locale_q_pair[0]: # Check for de-DE 

216 lang = locale_q_pair[0].split("-")[0] 

217 else: 

218 lang = locale_q_pair[0] 

219 if lang in conf.i18n.available_languages + list(conf.i18n.language_alias_map.keys()): 

220 return lang 

221 if lang == "*": # fallback 

222 return conf.i18n.available_languages[0] 

223 return None 

224 

225 if not conf.i18n.available_languages: 

226 # This project doesn't use the multi-language feature, nothing to do here 

227 return path 

228 if conf.i18n.language_method == "session": 

229 current_session = current.session.get() 

230 lang = conf.i18n.default_language 

231 # We save the language in the session, if it exists, and try to load it from there 

232 if "lang" in current_session: 

233 current.language.set(current_session["lang"]) 

234 return path 

235 

236 if header_lang := get_language_from_header(): 

237 lang = header_lang 

238 current.language.set(lang) 

239 

240 elif header_lang := self.request.headers.get("X-Appengine-Country"): 

241 header_lang = str(header_lang).lower() 

242 if header_lang in conf.i18n.available_languages + list(conf.i18n.language_alias_map.keys()): 

243 lang = header_lang 

244 

245 if current_session.loaded: 

246 current_session["lang"] = lang 

247 current.language.set(lang) 

248 

249 elif conf.i18n.language_method == "domain": 

250 host = self.request.host_url.lower() 

251 host = host[host.find("://") + 3:].strip(" /") # strip http(s):// 

252 if host.startswith("www."): 

253 host = host[4:] 

254 if lang := conf.i18n.domain_language_mapping.get(host): 

255 current.language.set(lang) 

256 # We have no language configured for this domain, try to read it from the HTTP Header 

257 elif lang := get_language_from_header(): 

258 current.language.set(lang) 

259 

260 elif conf.i18n.language_method == "url": 

261 tmppath = urlparse(path).path 

262 tmppath = [unquote(x) for x in tmppath.lower().strip("/").split("/")] 

263 if ( 

264 len(tmppath) > 0 

265 and tmppath[0] in conf.i18n.available_languages + list(conf.i18n.language_alias_map.keys()) 

266 ): 

267 current.language.set(tmppath[0]) 

268 return path[len(tmppath[0]) + 1:] # Return the path stripped by its language segment 

269 else: # This URL doesnt contain an language prefix, try to read it from session 

270 if header_lang := get_language_from_header(): 

271 current.language.set(header_lang) 

272 elif header_lang := self.request.headers.get("X-Appengine-Country"): 

273 lang = str(header_lang).lower() 

274 if lang in conf.i18n.available_languages or lang in conf.i18n.language_alias_map: 

275 current.language.set(lang) 

276 elif conf.i18n.language_method == "header": 

277 if lang := get_language_from_header(): 

278 current.language.set(lang) 

279 

280 return path 

281 

282 def _process(self): 

283 if self.method not in ("get", "post", "head", "options"): 

284 logging.error(f"{self.method=} not supported") 

285 return 

286 

287 if self.request.headers.get("X-AppEngine-TaskName", None) is not None: # Check if we run in the appengine 

288 if self.request.environ.get("HTTP_X_APPENGINE_USER_IP") in _appengineServiceIPs: 

289 self.is_deferred = True 

290 elif os.getenv("TASKS_EMULATOR") is not None: 

291 self.is_deferred = True 

292 

293 # Check if we should process or abort the request 

294 for validator, reqValidatorResult in [(x, x.validate(self)) for x in self.requestValidators]: 

295 if reqValidatorResult is not None: 

296 logging.warning(f"Request rejected by validator {validator.name}") 

297 statusCode, statusStr, statusDescr = reqValidatorResult 

298 self.response.status = f"{statusCode} {statusStr}" 

299 self.response.write(statusDescr) 

300 return 

301 

302 try: 

303 path = self.request.path 

304 except UnicodeDecodeError: # webob can fail with UnicodeDecodeError on broken/invalid URLs 

305 self.response.status = "400 Bad Request" # let's send the client onto a health cure in Bad Request ... 

306 return 

307 

308 # Add CSP headers early (if any) 

309 if conf.security.content_security_policy and conf.security.content_security_policy["_headerCache"]: 

310 for k, v in conf.security.content_security_policy["_headerCache"].items(): 

311 self.response.headers[k] = v 

312 # Endpoints referenced by the CSP-directive "report-to" and others 

313 if reporting_endpoints := _build_reporting_endpoints_header(): 

314 self.response.headers["Reporting-Endpoints"] = reporting_endpoints 

315 if self.isSSLConnection: # Check for HTST and PKP headers only if we have a secure channel. 

316 if conf.security.strict_transport_security: 

317 self.response.headers["Strict-Transport-Security"] = conf.security.strict_transport_security 

318 # Check for X-Security-Headers we shall emit 

319 if conf.security.x_content_type_options: 

320 self.response.headers["X-Content-Type-Options"] = "nosniff" 

321 if conf.security.x_xss_protection is not None: 

322 if conf.security.x_xss_protection: 

323 self.response.headers["X-XSS-Protection"] = "1; mode=block" 

324 elif conf.security.x_xss_protection is False: 

325 self.response.headers["X-XSS-Protection"] = "0" 

326 if conf.security.x_frame_options is not None and isinstance(conf.security.x_frame_options, tuple): 

327 mode, uri = conf.security.x_frame_options 

328 if mode in ["deny", "sameorigin"]: 

329 self.response.headers["X-Frame-Options"] = mode 

330 elif mode == "allow-from": 

331 self.response.headers["X-Frame-Options"] = f"allow-from {uri}" 

332 if conf.security.x_permitted_cross_domain_policies is not None: 

333 self.response.headers["X-Permitted-Cross-Domain-Policies"] = conf.security.x_permitted_cross_domain_policies 

334 if conf.security.referrer_policy: 

335 self.response.headers["Referrer-Policy"] = conf.security.referrer_policy 

336 if conf.security.permissions_policy.get("_headerCache"): 

337 self.response.headers["Permissions-Policy"] = conf.security.permissions_policy["_headerCache"] 

338 if conf.security.enable_coep: 

339 self.response.headers["Cross-Origin-Embedder-Policy"] = "require-corp" 

340 if conf.security.enable_coop: 

341 self.response.headers["Cross-Origin-Opener-Policy"] = conf.security.enable_coop 

342 if conf.security.enable_corp: 

343 self.response.headers["Cross-Origin-Resource-Policy"] = conf.security.enable_corp 

344 

345 # Ensure that TLS is used if required 

346 if conf.security.force_ssl and not self.isSSLConnection and not conf.instance.is_dev_server: 

347 isWhitelisted = False 

348 reqPath = self.request.path 

349 for testUrl in conf.security.no_ssl_check_urls: 

350 if testUrl.endswith("*"): 

351 if reqPath.startswith(testUrl[:-1]): 

352 isWhitelisted = True 

353 break 

354 else: 

355 if testUrl == reqPath: 

356 isWhitelisted = True 

357 break 

358 if not isWhitelisted: # Some URLs need to be whitelisted (as f.e. the Tasks-Queue doesn't call using https) 

359 # Redirect the user to the startpage (using ssl this time) 

360 host = self.request.host_url.lower() 

361 host = host[host.find("://") + 3:].strip(" /") # strip http(s):// 

362 self.response.status = "302 Found" 

363 self.response.headers['Location'] = f"https://{host}/" 

364 return 

365 if path.startswith("/_ah/warmup"): 

366 self.response.write("okay") 

367 return 

368 

369 try: 

370 current.session.get().load() 

371 

372 # Load current user into context variable if user module is there. 

373 if user_mod := getattr(conf.main_app.vi, "user", None): 

374 current.user.set(user_mod.getCurrentUser()) 

375 

376 path = self._select_language(path)[1:] 

377 

378 # Check for closed system 

379 if conf.security.closed_system and self.method != "options": 

380 if not current.user.get(): 

381 if not any(fnmatch.fnmatch(path, pat) for pat in conf.security.closed_system_allowed_paths): 

382 raise errors.Unauthorized() 

383 

384 if conf.request_preprocessor: 

385 path = conf.request_preprocessor(path) 

386 

387 self._route(path) 

388 

389 except errors.Redirect as e: 

390 if conf.debug.trace_exceptions: 

391 logging.warning("""conf.debug.trace_exceptions is set, won't handle this exception""") 

392 raise 

393 self.response.status = f"{e.status} {e.name}" 

394 url = e.url 

395 url = unquote(url) # decode first 

396 # safe = https://url.spec.whatwg.org/#url-path-segment-string 

397 url = quote(url, encoding="utf-8", safe="!$&'()*+,-./:;=?@_~#") # re-encode all in utf-8 

398 if url.startswith(('.', '/')): 

399 url = str(urljoin(self.request.url, url)) 

400 self.response.headers['Location'] = url 

401 

402 except Exception as e: 

403 if conf.debug.trace_exceptions: 

404 logging.warning("""conf.debug.trace_exceptions is set, won't handle this exception""") 

405 raise 

406 self.response.body = b"" 

407 if isinstance(e, errors.HTTPException): 

408 logging.info(f"[{e.status}] {e.name}: {e.descr}", exc_info=conf.debug.trace) 

409 self.response.status = f"{e.status} {e.name}" 

410 # Set machine-readable x-viur-error response header in case there is an exception description. 

411 if e.descr: 

412 self.response.headers["x-viur-error"] = e.descr.replace("\n", "") 

413 else: 

414 self.response.status = 500 

415 logging.error("ViUR has caught an unhandled exception!") 

416 logging.exception(e) 

417 

418 res = None 

419 if conf.error_handler: 

420 try: 

421 res = conf.error_handler(e) 

422 except Exception as newE: 

423 logging.error("viur.error_handler failed!") 

424 logging.exception(newE) 

425 res = None 

426 if not res: 

427 descr = "The server encountered an unexpected error and is unable to process your request." 

428 

429 if isinstance(e, errors.HTTPException): 

430 error_info = { 

431 "status": e.status, 

432 "reason": e.name, 

433 "title": str(translate(e.name)), 

434 "descr": e.descr, 

435 } 

436 else: 

437 error_info = { 

438 "status": 500, 

439 "reason": "Internal Server Error", 

440 "title": str(translate("Internal Server Error")), 

441 "descr": descr 

442 } 

443 

444 if conf.instance.is_dev_server: 

445 error_info["traceback"] = traceback.format_exc() 

446 

447 error_info["logo"] = conf.error_logo 

448 

449 if (len(self.path_list) > 0 and self.path_list[0] in ("vi", "json")) or \ 

450 current.request.get().response.headers["Content-Type"] == "application/json": 

451 current.request.get().response.headers["Content-Type"] = "application/json" 

452 res = json.dumps(error_info) 

453 else: # We render the error in html 

454 # Try to get the template from html/error/ 

455 if filename := conf.main_app.render.getTemplateFileName((f"{error_info['status']}", "error"), 

456 raise_exception=False): 

457 template = conf.main_app.render.getEnv().get_template(filename) 

458 try: 

459 uses_unsafe_inline = \ 

460 "unsafe-inline" in conf.security.content_security_policy["enforce"]["style-src"] 

461 except (KeyError, TypeError): # Not set 

462 uses_unsafe_inline = False 

463 if uses_unsafe_inline: 

464 logging.info("Using style-src:unsafe-inline, don't create a nonce") 

465 nonce = None 

466 else: 

467 nonce = utils.string.random(16) 

468 extendCsp({"style-src": [f"nonce-{nonce}"]}) 

469 res = template.render(error_info, nonce=nonce) 

470 else: 

471 res = (f'<!DOCTYPE html><html lang="en"><head><meta charset="UTF-8">' 

472 f'<title>{error_info["status"]} - {error_info["reason"]}</title>' 

473 f'</head><body><h1>{error_info["status"]} - {error_info["reason"]}</h1>') 

474 

475 self.response.write(res.encode("UTF-8")) 

476 

477 finally: 

478 current.session.get().save() 

479 if conf.instance.is_dev_server and conf.debug.dev_server_cloud_logging: 

480 # Emit the outer log only on dev_appserver (we'll use the existing request log when live) 

481 SEVERITY = "DEBUG" 

482 if self.maxLogLevel >= 50: 

483 SEVERITY = "CRITICAL" 

484 elif self.maxLogLevel >= 40: 

485 SEVERITY = "ERROR" 

486 elif self.maxLogLevel >= 30: 

487 SEVERITY = "WARNING" 

488 elif self.maxLogLevel >= 20: 

489 SEVERITY = "INFO" 

490 

491 TRACE = "projects/{}/traces/{}".format(loggingClient.project, self._traceID) 

492 

493 REQUEST = { 

494 'requestMethod': self.request.method, 

495 'requestUrl': self.request.url, 

496 'status': self.response.status_code, 

497 'userAgent': self.request.headers.get('USER-AGENT'), 

498 'responseSize': self.response.content_length, 

499 'latency': "%0.3fs" % (time.time() - self.startTime), 

500 'remoteIp': self.request.environ.get("HTTP_X_APPENGINE_USER_IP") 

501 } 

502 requestLogger.log_text( 

503 "", 

504 client=loggingClient, 

505 severity=SEVERITY, 

506 http_request=REQUEST, 

507 trace=TRACE, 

508 resource=requestLoggingRessource, 

509 operation={ 

510 "first": True, 

511 "last": True, 

512 "id": self._traceID 

513 } 

514 ) 

515 

516 if conf.instance.is_dev_server: 

517 self.is_deferred = True 

518 

519 while self.pendingTasks: 

520 task = self.pendingTasks.pop() 

521 logging.debug(f"Deferred task emulation, executing {task=}") 

522 try: 

523 task() 

524 except Exception: # noqa 

525 logging.exception(f"Deferred Task emulation {task} failed") 

526 

527 def _route(self, path: str) -> None: 

528 """ 

529 Does the actual work of sanitizing the parameter, determine which exposed-function to call 

530 (and with which parameters) 

531 """ 

532 

533 # Parse the URL 

534 if path := parse.urlparse(path).path: 

535 self.path = path 

536 self.path_list = tuple(unicodedata.normalize("NFC", parse.unquote(part)) 

537 for part in path.strip("/").split("/")) 

538 

539 # Prevent Hash-collision attacks 

540 if len(self.request.params) > conf.max_post_params_count: 

541 raise errors.BadRequest( 

542 f"Too many arguments supplied, exceeding maximum" 

543 f" of {conf.max_post_params_count} allowed arguments per request" 

544 ) 

545 

546 param_filter = conf.param_filter_function 

547 if param_filter and not callable(param_filter): 

548 raise ValueError(f"""{param_filter=} is not callable""") 

549 

550 for key, value in self.request.params.items(): 

551 try: 

552 key = unicodedata.normalize("NFC", key) 

553 value = unicodedata.normalize("NFC", value) 

554 except UnicodeError: 

555 # We received invalid unicode data (usually happens when 

556 # someone tries to exploit unicode normalisation bugs) 

557 raise errors.BadRequest() 

558 

559 if param_filter and param_filter(key, value): 

560 continue 

561 

562 if key == TEMPLATE_STYLE_KEY: 

563 self.template_style = value 

564 continue 

565 

566 if key in self.kwargs: 

567 if isinstance(self.kwargs[key], list): 

568 self.kwargs[key].append(value) 

569 else: # Convert that key to a list 

570 self.kwargs[key] = [self.kwargs[key], value] 

571 else: 

572 self.kwargs[key] = value 

573 

574 if "self" in self.kwargs or "return" in self.kwargs: # self or return is reserved for bound methods 

575 raise errors.BadRequest() 

576 

577 caller = conf.main_resolver 

578 idx = 0 # Count how may items from *args we'd have consumed (so the rest can go into *args of the called func 

579 path_found = True 

580 

581 for part in self.path_list: 

582 # TODO: Remove canAccess guards... solve differently. 

583 if "canAccess" in caller and not caller["canAccess"](): 

584 # We have a canAccess function guarding that object, 

585 # and it returns False... 

586 raise errors.Unauthorized() 

587 

588 idx += 1 

589 

590 if part not in caller: 

591 part = "index" 

592 

593 if caller := caller.get(part): 

594 if isinstance(caller, Method): 

595 if part == "index": 

596 idx -= 1 

597 

598 self.args = tuple(self.path_list[idx:]) 

599 break 

600 

601 elif part == "index": 

602 path_found = False 

603 break 

604 

605 else: 

606 path_found = False 

607 break 

608 

609 if not path_found: 

610 raise errors.NotFound( 

611 f"""The path {utils.string.escape("/".join(self.path_list[:idx]))} could not be found""") 

612 

613 if not isinstance(caller, Method): 

614 # try to find "index" function 

615 if (index := caller.get("index")) and isinstance(index, Method): 

616 caller = index 

617 else: 

618 raise errors.MethodNotAllowed() 

619 

620 # Check for internal exposed 

621 if caller.exposed is False and not self.internalRequest: 

622 raise errors.NotFound() 

623 

624 # Fill the Allow header of the response with the allowed HTTP methods 

625 if self.method == "options": 

626 self.response.headers["Allow"] = ", ".join(sorted(caller.methods)).upper() 

627 

628 # Register caller specific CORS headers 

629 self.cors_headers = [str(header).lower() for header in caller.cors_allow_headers or ()] 

630 

631 # Check for @force_ssl flag 

632 if not self.internalRequest \ 

633 and caller.ssl \ 

634 and not self.request.host_url.lower().startswith("https://") \ 

635 and not conf.instance.is_dev_server: 

636 raise errors.PreconditionFailed("You must use SSL to access this resource!") 

637 

638 # Check for @force_post flag 

639 if not self.isPostRequest and caller.methods == ("POST",): 

640 raise errors.MethodNotAllowed("You must use POST to access this resource!") 

641 

642 # Check if this request should bypass the caches 

643 if self.request.headers.get("X-Viur-Disable-Cache"): 

644 # No cache requested, check if the current user is allowed to do so 

645 if (user := current.user.get()) and "root" in user["access"]: 

646 logging.debug("Caching disabled by X-Viur-Disable-Cache header") 

647 self.disableCache = True 

648 

649 # Destill context as self.context, if available 

650 if context := {k: v for k, v in self.kwargs.items() if k.startswith("@")}: 

651 # Remove context parameters from kwargs 

652 kwargs = {k: v for k, v in self.kwargs.items() if k not in context} 

653 # Remove leading "@" from context parameters 

654 self.context |= {k[1:]: v for k, v in context.items() if len(k) > 1} 

655 else: 

656 kwargs = self.kwargs 

657 

658 if ((self.internalRequest and conf.debug.trace_internal_call_routing) 

659 or conf.debug.trace_external_call_routing): 

660 logging.debug( 

661 f"Calling {caller._func!r} with args={self.args!r}, {kwargs=} within context={self.context!r}" 

662 ) 

663 

664 if self.method == "options": 

665 # OPTIONS request doesn't have a body 

666 del self.response.app_iter 

667 del self.response.content_type 

668 self.response.status = "204 No Content" 

669 return 

670 

671 # Now call the routed method! 

672 res = caller(*self.args, **kwargs) 

673 

674 if self.method == "options": 

675 # OPTIONS request doesn't have a body 

676 del self.response.app_iter 

677 del self.response.content_type 

678 self.response.status = "204 No Content" 

679 return 

680 

681 if not isinstance(res, bytes): # Convert the result to bytes if it is not already! 

682 res = str(res).encode("UTF-8") 

683 self.response.write(res) 

684 

685 def _cors(self) -> None: 

686 """ 

687 Set CORS headers to the HTTP response. 

688 

689 .. seealso:: 

690 

691 Option :attr:`core.config.Security.cors_origins`, etc. 

692 for cors settings. 

693 

694 https://fetch.spec.whatwg.org/#http-cors-protocol 

695 

696 https://enable-cors.org/server.html 

697 

698 https://www.html5rocks.com/static/images/cors_server_flowchart.png 

699 """ 

700 

701 def test_candidates(value: str, *candidates: str | re.Pattern) -> bool: 

702 """Test if the value matches the pattern of any candidate""" 

703 for candidate in candidates: 

704 if isinstance(candidate, re.Pattern): 

705 if candidate.match(value): 

706 return True 

707 elif isinstance(candidate, str): 

708 if candidate.lower() == str(value).lower(): 

709 return True 

710 else: 

711 raise TypeError( 

712 f"Invalid setting {candidate}. " 

713 f"Expected a string or a compiled regex." 

714 ) 

715 return False 

716 

717 origin = current.request.get().request.headers.get("Origin") 

718 if not origin: 

719 return 

720 

721 # Origin is set --> It's a CORS request 

722 

723 any_origin_allowed = ( 

724 conf.security.cors_origins == "*" 

725 or any(_origin == "*" for _origin in conf.security.cors_origins) 

726 or any(_origin.pattern == r".*" 

727 for _origin in conf.security.cors_origins 

728 if isinstance(_origin, re.Pattern)) 

729 ) 

730 

731 if any_origin_allowed and conf.security.cors_origins_use_wildcard: 

732 if conf.security.cors_allow_credentials: 

733 raise RuntimeError( 

734 "Invalid CORS config: " 

735 "If credentials mode is \"include\", then `Access-Control-Allow-Origin` cannot be `*`. " 

736 "See https://fetch.spec.whatwg.org/#cors-protocol-and-credentials" 

737 ) 

738 self.response.headers["Access-Control-Allow-Origin"] = "*" 

739 

740 elif test_candidates(origin, *conf.security.cors_origins): 

741 self.response.headers["Access-Control-Allow-Origin"] = origin 

742 

743 else: 

744 logging.warning(f"{origin=} not valid (must be one of {conf.security.cors_origins=})") 

745 return 

746 

747 if conf.security.cors_allow_credentials: