Store a managed session
(self, session)
| 399 | ) |
| 400 | |
| 401 | def put(self, session): |
| 402 | """Store a managed session""" |
| 403 | current_time = time.time() |
| 404 | if not session.hmac_digest: |
| 405 | session.sign(self.secret) |
| 406 | elif not session.force_write and session.last_write is not None and \ |
| 407 | (current_time - float(session.last_write)) < \ |
| 408 | self.disk_write_delay: |
| 409 | return |
| 410 | |
| 411 | session.last_write = current_time |
| 412 | session.force_write = False |
| 413 | |
| 414 | # Do not store the session if skip paths |
| 415 | for sp in self.skip_paths: |
| 416 | if request.path.startswith(sp): |
| 417 | return |
| 418 | |
| 419 | fname = safe_join(self.path, session.sid) |
| 420 | |
| 421 | if fname is None: |
| 422 | raise InternalServerError('Failed to update the session') |
| 423 | |
| 424 | plaintext = pickle.dumps( |
| 425 | (session.randval, session.hmac_digest, dict(session)), -1 |
| 426 | ) |
| 427 | # Encrypt-then-MAC: Fernet wrap first, then HMAC over the |
| 428 | # ciphertext. pickle.loads on a tampered or unauthenticated body |
| 429 | # is therefore unreachable on the read path. |
| 430 | body = _derive_session_fernet(self.secret).encrypt(plaintext) |
| 431 | header = _compute_file_hmac(self.secret, body) |
| 432 | with _open_session_file(fname) as f: |
| 433 | f.write(header) |
| 434 | f.write(body) |
| 435 | |
| 436 | |
| 437 | class ManagedSessionInterface(SessionInterface): |
nothing calls this directly
no test coverage detected