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 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 - Normalize & sanity check the parameters 

110 - Resolve the exposed function and call it 

111 - Save the session / tear down the request 

112 - Return the response generated 

113 

114 

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

116 """ 

117 

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

119 requestValidators = [FetchMetaDataValidator] 

120 

121 def __init__(self, environ: dict): 

122 super().__init__() 

123 self.startTime = time.time() 

124 

125 self.request = webob.Request(environ) 

126 self.response = webob.Response() 

127 

128 self.maxLogLevel = logging.DEBUG 

129 self._traceID = \ 

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

131 self.is_deferred = False 

132 self.path = "" 

133 self.path_list = () 

134 

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

136 self.internalRequest = False 

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

138 self.pendingTasks = [] 

139 self.args = () 

140 self.kwargs = {} 

141 self.context = {} 

142 self.template_style: str | None = None 

143 self.cors_headers = () 

144 

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

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

147 self.isPostRequest = self.method == "post" 

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

149 

150 db.current_db_access_log.set(set()) 

151 

152 # Set context variables 

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

154 current.request.set(self) 

155 current.session.set(session.Session()) 

156 current.request_data.set({}) 

157 

158 # Process actual request 

159 self._process() 

160 

161 self._cors() 

162 

163 # Unset context variables 

164 current.language.set(None) 

165 current.request_data.set(None) 

166 current.session.set(None) 

167 current.request.set(None) 

168 current.user.set(None) 

169 

170 @property 

171 def isDevServer(self) -> bool: 

172 import warnings 

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

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

175 logging.warning(msg) 

176 return conf.instance.is_dev_server 

177 

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

179 """ 

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

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

182 or extract it from the URL. 

183 """ 

184 

185 def get_language_from_header() -> str | None: 

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

187 return None 

188 languages = accept_language.split(",") 

189 locale_q_pairs = [] 

190 

191 for language in languages: 

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

193 # no q => q = 1 

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

195 else: 

196 try: 

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

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

199 locale_q_pairs.append((locale, q)) 

200 except IndexError: 

201 continue # skip language 

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

203 for locale_q_pair in locale_q_pairs: 

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

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

206 else: 

207 lang = locale_q_pair[0] 

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

209 return lang 

210 if lang == "*": # fallback 

211 return conf.i18n.available_languages[0] 

212 return None 

213 

214 if not conf.i18n.available_languages: 

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

216 return path 

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

218 current_session = current.session.get() 

219 lang = conf.i18n.default_language 

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

221 if "lang" in current_session: 

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

223 return path 

224 

225 if header_lang := get_language_from_header(): 

226 lang = header_lang 

227 current.language.set(lang) 

228 

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

230 header_lang = str(header_lang).lower() 

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

232 lang = header_lang 

233 

234 if current_session.loaded: 

235 current_session["lang"] = lang 

236 current.language.set(lang) 

237 

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

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

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

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

242 host = host[4:] 

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

244 current.language.set(lang) 

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

246 elif lang := get_language_from_header(): 

247 current.language.set(lang) 

248 

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

250 tmppath = urlparse(path).path 

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

252 if ( 

253 len(tmppath) > 0 

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

255 ): 

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

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

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

259 if header_lang := get_language_from_header(): 

260 current.language.set(header_lang) 

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

262 lang = str(header_lang).lower() 

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

264 current.language.set(lang) 

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

266 if lang := get_language_from_header(): 

267 current.language.set(lang) 

268 

269 return path 

270 

271 def _process(self): 

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

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

274 return 

275 

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

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

278 self.is_deferred = True 

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

280 self.is_deferred = True 

281 

282 # Check if we should process or abort the request 

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

284 if reqValidatorResult is not None: 

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

286 statusCode, statusStr, statusDescr = reqValidatorResult 

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

288 self.response.write(statusDescr) 

289 return 

290 

291 try: 

292 path = self.request.path 

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

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

295 return 

296 

297 # Add CSP headers early (if any) 

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

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

300 self.response.headers[k] = v 

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

302 if conf.security.strict_transport_security: 

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

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

305 if conf.security.x_content_type_options: 

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

307 if conf.security.x_xss_protection is not None: 

308 if conf.security.x_xss_protection: 

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

310 elif conf.security.x_xss_protection is False: 

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

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

313 mode, uri = conf.security.x_frame_options 

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

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

316 elif mode == "allow-from": 

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

318 if conf.security.x_permitted_cross_domain_policies is not None: 

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

320 if conf.security.referrer_policy: 

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

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

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

324 if conf.security.enable_coep: 

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

326 if conf.security.enable_coop: 

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

328 if conf.security.enable_corp: 

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

330 

331 # Ensure that TLS is used if required 

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

333 isWhitelisted = False 

334 reqPath = self.request.path 

335 for testUrl in conf.security.no_ssl_check_urls: 

336 if testUrl.endswith("*"): 

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

338 isWhitelisted = True 

339 break 

340 else: 

341 if testUrl == reqPath: 

342 isWhitelisted = True 

343 break 

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

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

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

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

348 self.response.status = "302 Found" 

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

350 return 

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

352 self.response.write("okay") 

353 return 

354 

355 try: 

356 current.session.get().load() 

357 

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

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

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

361 

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

363 

364 # Check for closed system 

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

366 if not current.user.get(): 

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

368 raise errors.Unauthorized() 

369 

370 if conf.request_preprocessor: 

371 path = conf.request_preprocessor(path) 

372 

373 self._route(path) 

374 

375 except errors.Redirect as e: 

376 if conf.debug.trace_exceptions: 

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

378 raise 

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

380 url = e.url 

381 url = unquote(url) # decode first 

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

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

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

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

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

387 

388 except Exception as e: 

389 if conf.debug.trace_exceptions: 

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

391 raise 

392 self.response.body = b"" 

393 if isinstance(e, errors.HTTPException): 

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

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

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

397 if e.descr: 

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

399 else: 

400 self.response.status = 500 

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

402 logging.exception(e) 

403 

404 res = None 

405 if conf.error_handler: 

406 try: 

407 res = conf.error_handler(e) 

408 except Exception as newE: 

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

410 logging.exception(newE) 

411 res = None 

412 if not res: 

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

414 

415 if isinstance(e, errors.HTTPException): 

416 error_info = { 

417 "status": e.status, 

418 "reason": e.name, 

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

420 "descr": e.descr, 

421 } 

422 else: 

423 error_info = { 

424 "status": 500, 

425 "reason": "Internal Server Error", 

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

427 "descr": descr 

428 } 

429 

430 if conf.instance.is_dev_server: 

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

432 

433 error_info["logo"] = conf.error_logo 

434 

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

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

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

438 res = json.dumps(error_info) 

439 else: # We render the error in html 

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

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

442 raise_exception=False): 

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

444 try: 

445 uses_unsafe_inline = \ 

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

447 except (KeyError, TypeError): # Not set 

448 uses_unsafe_inline = False 

449 if uses_unsafe_inline: 

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

451 nonce = None 

452 else: 

453 nonce = utils.string.random(16) 

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

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

456 else: 

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

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

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

460 

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

462 

463 finally: 

464 current.session.get().save() 

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

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

467 SEVERITY = "DEBUG" 

468 if self.maxLogLevel >= 50: 

469 SEVERITY = "CRITICAL" 

470 elif self.maxLogLevel >= 40: 

471 SEVERITY = "ERROR" 

472 elif self.maxLogLevel >= 30: 

473 SEVERITY = "WARNING" 

474 elif self.maxLogLevel >= 20: 

475 SEVERITY = "INFO" 

476 

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

478 

479 REQUEST = { 

480 'requestMethod': self.request.method, 

481 'requestUrl': self.request.url, 

482 'status': self.response.status_code, 

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

484 'responseSize': self.response.content_length, 

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

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

487 } 

488 requestLogger.log_text( 

489 "", 

490 client=loggingClient, 

491 severity=SEVERITY, 

492 http_request=REQUEST, 

493 trace=TRACE, 

494 resource=requestLoggingRessource, 

495 operation={ 

496 "first": True, 

497 "last": True, 

498 "id": self._traceID 

499 } 

500 ) 

501 

502 if conf.instance.is_dev_server: 

503 self.is_deferred = True 

504 

505 while self.pendingTasks: 

506 task = self.pendingTasks.pop() 

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

508 try: 

509 task() 

510 except Exception: # noqa 

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

512 

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

514 """ 

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

516 (and with which parameters) 

517 """ 

518 

519 # Parse the URL 

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

521 self.path = path 

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

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

524 

525 # Prevent Hash-collision attacks 

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

527 raise errors.BadRequest( 

528 f"Too many arguments supplied, exceeding maximum" 

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

530 ) 

531 

532 param_filter = conf.param_filter_function 

533 if param_filter and not callable(param_filter): 

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

535 

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

537 try: 

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

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

540 except UnicodeError: 

541 # We received invalid unicode data (usually happens when 

542 # someone tries to exploit unicode normalisation bugs) 

543 raise errors.BadRequest() 

544 

545 if param_filter and param_filter(key, value): 

546 continue 

547 

548 if key == TEMPLATE_STYLE_KEY: 

549 self.template_style = value 

550 continue 

551 

552 if key in self.kwargs: 

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

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

555 else: # Convert that key to a list 

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

557 else: 

558 self.kwargs[key] = value 

559 

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

561 raise errors.BadRequest() 

562 

563 caller = conf.main_resolver 

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

565 path_found = True 

566 

567 for part in self.path_list: 

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

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

570 # We have a canAccess function guarding that object, 

571 # and it returns False... 

572 raise errors.Unauthorized() 

573 

574 idx += 1 

575 

576 if part not in caller: 

577 part = "index" 

578 

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

580 if isinstance(caller, Method): 

581 if part == "index": 

582 idx -= 1 

583 

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

585 break 

586 

587 elif part == "index": 

588 path_found = False 

589 break 

590 

591 else: 

592 path_found = False 

593 break 

594 

595 if not path_found: 

596 raise errors.NotFound( 

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

598 

599 if not isinstance(caller, Method): 

600 # try to find "index" function 

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

602 caller = index 

603 else: 

604 raise errors.MethodNotAllowed() 

605 

606 # Check for internal exposed 

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

608 raise errors.NotFound() 

609 

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

611 if self.method == "options": 

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

613 

614 # Register caller specific CORS headers 

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

616 

617 # Check for @force_ssl flag 

618 if not self.internalRequest \ 

619 and caller.ssl \ 

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

621 and not conf.instance.is_dev_server: 

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

623 

624 # Check for @force_post flag 

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

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

627 

628 # Check if this request should bypass the caches 

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

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

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

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

633 self.disableCache = True 

634 

635 # Destill context as self.context, if available 

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

637 # Remove context parameters from kwargs 

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

639 # Remove leading "@" from context parameters 

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

641 else: 

642 kwargs = self.kwargs 

643 

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

645 or conf.debug.trace_external_call_routing): 

646 logging.debug( 

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

648 ) 

649 

650 if self.method == "options": 

651 # OPTIONS request doesn't have a body 

652 del self.response.app_iter 

653 del self.response.content_type 

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

655 return 

656 

657 # Now call the routed method! 

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

659 

660 if self.method == "options": 

661 # OPTIONS request doesn't have a body 

662 del self.response.app_iter 

663 del self.response.content_type 

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

665 return 

666 

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

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

669 self.response.write(res) 

670 

671 def _cors(self) -> None: 

672 """ 

673 Set CORS headers to the HTTP response. 

674 

675 .. seealso:: 

676 

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

678 for cors settings. 

679 

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

681 

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

683 

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

685 """ 

686 

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

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

689 for candidate in candidates: 

690 if isinstance(candidate, re.Pattern): 

691 if candidate.match(value): 

692 return True 

693 elif isinstance(candidate, str): 

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

695 return True 

696 else: 

697 raise TypeError( 

698 f"Invalid setting {candidate}. " 

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

700 ) 

701 return False 

702 

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

704 if not origin: 

705 return 

706 

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

708 

709 any_origin_allowed = ( 

710 conf.security.cors_origins == "*" 

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

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

713 for _origin in conf.security.cors_origins 

714 if isinstance(_origin, re.Pattern)) 

715 ) 

716 

717 if any_origin_allowed and conf.security.cors_origins_use_wildcard: 

718 if conf.security.cors_allow_credentials: 

719 raise RuntimeError( 

720 "Invalid CORS config: " 

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

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

723 ) 

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

725 

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

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

728 

729 else: 

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

731 return 

732 

733 if conf.security.cors_allow_credentials: 

734 self.response.headers["Access-Control-Allow-Credentials"] = "true" 

735 

736 if self.method == "options": 

737 method = (self.request.headers.get("Access-Control-Request-Method") or "").lower() 

738 

739 if method in conf.security.cors_methods: 

740 # It's a CORS-preflight request 

741 # - MUST include Access-Control-Request-Method 

742 # - CAN include Access-Control-Request-Headers 

743 

744 # The response can be cached 

745 if conf.security.cors_max_age is not None: 

746 assert isinstance(conf.security.cors_max_age, datetime.timedelta) 

747 self.response.headers["Access-Control-Max-Age"] = \