0001-misc-use-database-to-store-form-tokens-71455.patch

Frédéric Péters, 08 décembre 2022 22:05

Voir les différences: en ligne côte à côte

Subject: [PATCH] misc: use database to store form tokens (#71455)

 tests/test_sessions.py  |  8 ++++---
 wcs/qommon/publisher.py | 19 -----------------
 wcs/qommon/sessions.py  | 46 +----------------------------------------
 wcs/sql.py              | 18 +++++++++++++---
 4 files changed, 21 insertions(+), 70 deletions(-)

         resp = resp.form.submit('submit')
         assert sql.Session.count() == 1
         assert sql.TransientData.count() == 1
         transient_data = sql.TransientData.select()[0]
         assert sql.TransientData.count() == 2
         transient_data1, transient_data2 = sql.TransientData.select(order_by='last_update_time')
         assert transient_data1.data is None  # form_token
         assert transient_data2.data == {'1': 'test'}  # magictoken
         app.get('/logout')
         assert sql.Session.count() == 0
         assert sql.TransientData.count() == 0
         transient_data.store()  # session_id not found, should not fail
         transient_data2.store()  # session_id not found, should not fail
     def test_magictoken_migration(pub, app):

         app_dir = None
         _i18n_catalog = None
         @property
         def form_tokens_dir(self):
             return os.path.join(self.app_dir, 'form_tokens')
         def get_root_url(self):
             if self.get_request():
                 return self.get_request().environ['SCRIPT_NAME'] + '/'
-...
                     raise ImmediateRedirectException(self.missing_appdir_redirect)
                 raise Http404()
             try:
                 os.mkdir(self.form_tokens_dir)
             except OSError:  # already exists
                 pass
         def init_publish(self, request):
             self.set_app_dir(request)
-...
                             except KeyError:
                                 pass
                             continue
                     # also delete obsolete form_tokens that would have be missed when
                     # cleaning sessions.
                     form_tokens_dir = self.form_tokens_dir
                     if os.path.exists(form_tokens_dir):
                         for filename in os.listdir(form_tokens_dir):
                             try:
                                 if os.stat(os.path.join(form_tokens_dir, filename)).st_mtime < creation_limit:
                                     os.unlink(os.path.join(form_tokens_dir, filename))
                             except FileNotFoundError:
                                 pass
             except locket.LockError:
                 pass

         session_id = property(get_session_id, set_session_id)
         def get_form_token_filepath(self, token):
             return os.path.join(get_publisher().form_tokens_dir, token)
         def create_form_token(self):
             token = super().create_form_token()
             with open(self.get_form_token_filepath(token), 'wb'):
                 # create empty file
                 pass
             return token
         def has_form_token(self, token):
             if not token:
                 return False
             has_form_token = super().has_form_token(token)
             if not os.path.exists(self.get_form_token_filepath(token)):
                 has_form_token = False
             return has_form_token
         def remove_form_token(self, token):
             super().remove_form_token(token)
             self.store()
             try:
                 os.unlink(self.get_form_token_filepath(token))
             except OSError:
                 pass
         def clean_form_tokens(self):
             dirname = os.path.join(get_publisher().app_dir, 'form_tokens')
             for token in self._form_tokens:
                 try:
                     os.unlink(os.path.join(dirname, token))
                 except OSError:
                     pass
         def has_user(self):
             user_id = QuixoteSession.get_user(self)
             return bool(user_id)
-...
         def __delitem__(self, session_id):
             if not session_id:
                 return
             try:
                 session = self.session_class.get(session_id)
             except KeyError:
                 session = None
             try:
                 self.session_class.remove_object(session_id)
             except OSError:
                 raise KeyError
             if session:
                 session.clean_form_tokens()
             self.session_class.remove_object(session_id)
         def get_sessions_for_saml(self, name_identifier=Ellipsis, session_indexes=()):
             return self.session_class.get_sessions_for_saml(name_identifier, session_indexes)

     import itertools
     import json
     import re
     import secrets
     import time
     import psycopg2
-...
     class TransientData(SqlMixin):
         # table to keep some transient submission data out of global session dictionary
         # table to keep some transient submission data and form tokens out of global session dictionary
         _table_name = 'transient_data'
         _table_static_fields = [
             ('id', 'varchar'),
-...
             sql_dict = {
                 'id': self.id,
                 'session_id': self.session_id,
                 'data': bytearray(pickle.dumps(self.data, protocol=2)),
                 'data': bytearray(pickle.dumps(self.data, protocol=2)) if self.data is not None else None,
                 'last_update_time': now(),
+            }
-...
             o = cls.__new__(cls)
             o.id = str_encode(row[0])
             o.session_id = row[1]
             o.data = pickle_loads(row[2])
             o.data = pickle_loads(row[2]) if row[2] else None
             return o
         @classmethod
-...
             super().remove_magictoken(token)
             TransientData.remove_object(token)
         def create_form_token(self):
             token = TransientData(id=secrets.token_urlsafe(16), session_id=self.id, data=None)
             token.store()
             return token.id
         def has_form_token(self, token):
             return TransientData.exists([Equal('id', token)])
         def remove_form_token(self, token):
             TransientData.remove_object(token)
     class TrackingCode(SqlMixin, wcs.tracking_code.TrackingCode):
         _table_name = 'tracking_codes'
+    -

Projet

Général

Profil

Produits Entr'ouvert » w.c.s.

0001-misc-use-database-to-store-form-tokens-71455.patch