[IMP] session_db: Modify the way to parse data from session data

- Add a way to detect the binary values to parse as strings when saving the session to the database. - To identify the data that was converted from binary to str, add a prefix base64:: to the string. - Then, when the session needs to be read, detect the values modified, and parse again to base64. - Verify if some value of the session is a monetary value and parse it to a float type. - Keep string type in the value of the 'debug' website parameter.

Author: JonathanOsAlc

session_db/pg_session_store.py

@@ -2,9 +2,11 @@
 # @author Nicolas Seinlet
 # Copyright (c) ACSONE SA 2022
 # @author Stéphane Bidoul
+import base64
 import json
 import logging
 import os
+import re
 
 import psycopg2
 
@@ -66,6 +68,7 @@ def __init__(self, uri, session_class=None):
  self._cr = None
  self._open_connection()
  self._setup_db()
+ self.prefix_binary = "base64::"
 
  def __del__(self):
  self._close_connection()
@@ -108,7 +111,8 @@ def _setup_db(self):
  @with_lock
  @with_cursor
  def save(self, session):
- payload = json.dumps(dict(session))
+ json_session = self.session_to_str(dict(session))
+ payload = json.dumps(json_session)
  self._cr.execute(
  """
  INSERT INTO http_sessions(sid, write_date, payload)
@@ -131,6 +135,7 @@ def get(self, sid):
  self._cr.execute("SELECT payload FROM http_sessions WHERE sid=%s", (sid,))
  try:
  data = json.loads(self._cr.fetchone()[0])
+ data = self.str_to_session(data)
  except Exception:
  return self.new()
 
@@ -149,6 +154,73 @@ def vacuum(self, max_lifetime=http.SESSION_LIFETIME):
  (f"{max_lifetime} seconds",),
  )
 
+ def _traverse_and_convert(self, data_node, conversion_func):
+ """Helper method that preserves keys while converting values."""
+ if isinstance(data_node, dict):
+ res = {}
+ for key, value in data_node.items():
+ # This is necessary because Odoo's core (ir_qweb) needs the 'debug' value as
+ # a string.
+ # The value for this key can be: "1", "assets", "True", "False", etc.
+ # Ref: https://github.com/Vauxoo/odoo/blob/d4d64d613800b8dc44c3262e13a2a81dbf3c742c/
+ # odoo/addons/base/models/ir_qweb.py#L912
+ # A test on an Odoo instance without the 'session_db' module confirmed
+ # that 'request.session.debug' value is always a string (str) type.
+ if key != "debug":
+ key = self._traverse_and_convert(key, conversion_func)
+ value = self._traverse_and_convert(value, conversion_func)
+ res.update({key: value})
+ return res
+ if isinstance(data_node, list):
+ return [
+ self._traverse_and_convert(item, conversion_func) for item in data_node
+ ]
+ return conversion_func(data_node)
+
+ def session_to_str(self, data):
+ """Converts binary values to prefixed strings."""
+
+ def convert(value):
+ if isinstance(value, bytes):
+ base64_string = base64.b64encode(value).decode("utf-8")
+ return self.prefix_binary + base64_string
+ return value
+
+ return self._traverse_and_convert(data, convert)
+
+ def str_to_session(self, data):
+ """Converts binary str to binary value again.
+ Converts int/float str values convert to their respective types.
+ """
+
+ def convert(value):
+ if not isinstance(value, str):
+ return value # Only process strings
+ # 1. Check for binary
+ if value.startswith(self.prefix_binary):
+ base64_string = value[len(self.prefix_binary) :]
+ try:
+ return base64.b64decode(base64_string)
+ except (ValueError, TypeError):
+ pass
+ numeric_parsers = [
+ # 2. Check for float (positive or negative)
+ # This regex requires a decimal point.
+ (r"^-?\d+\.\d+$", float),
+ # 3. Check for integer (positive or negative)
+ # This regex matches only digits (with optional sign).
+ (r"^-?\d+$", int),
+ ]
+ for pattern, parser in numeric_parsers:
+ if re.match(pattern, value):
+ try:
+ return parser(value)
+ except (ValueError, TypeError):
+ pass
+ return value
+
+ return self._traverse_and_convert(data, convert)
+
 
 _original_session_store = http.root.__class__.session_store
 

session_db/tests/test_pg_session_store.py

@@ -1,3 +1,4 @@
+import base64
 from unittest import mock
 
 import psycopg2
@@ -92,3 +93,104 @@ def test_make_postgres_uri(self):
  assert "postgres://test:PASSWORD@localhost:5432/test" == _make_postgres_uri(
  **connection_info
  )
+
+ def test_binary_serialization_roundtrip(self):
+ """Tests that binary data (bytes) converts to a string with a prefix
+ when saving, and converts back to bytes when reading.
+ """
+ original_data = {
+ "normal_text": "test",
+ "binary_data": b"Test binary",
+ }
+
+ # 1. Simulate save (session_to_str)
+ serialized = self.session_store.session_to_str(original_data)
+ # Verify that the prefix was added and encoded in base64
+ expected_b64 = base64.b64encode(b"Test binary").decode("utf-8")
+ self.assertEqual(
+ serialized["binary_data"],
+ f"base64::{expected_b64}",
+ "Binary data should be serialized with the base64:: prefix",
+ )
+ self.assertEqual(serialized["normal_text"], "test")
+
+ # 2. Simulate read (str_to_session)
+ deserialized = self.session_store.str_to_session(serialized)
+ # Verify that we recover the 'bytes' type
+ self.assertEqual(deserialized["binary_data"], b"Test binary")
+ self.assertIsInstance(deserialized["binary_data"], bytes)
+
+ def test_numeric_conversion(self):
+ """Tests that strings looking like numbers convert to int/float."""
+ data_from_json = {
+ "integer_str": "42",
+ "float_str": "3.1416",
+ "negative_int": "-10",
+ "negative_float": "-0.01",
+ "plain_text": "123-abc", # Should not convert
+ }
+ result = self.session_store.str_to_session(data_from_json)
+ # Integer validations
+ self.assertEqual(result["integer_str"], 42)
+ self.assertIsInstance(result["integer_str"], int)
+ self.assertEqual(result["negative_int"], -10)
+ # Float validations
+ self.assertEqual(result["float_str"], 3.1416)
+ self.assertIsInstance(result["float_str"], float)
+ self.assertEqual(result["negative_float"], -0.01)
+ # Text validations that must not change
+ self.assertEqual(result["plain_text"], "123-abc")
+ self.assertIsInstance(result["plain_text"], str)
+
+ def test_debug_param_exception(self):
+ """Verifies that the "debug" key ALWAYS remains a string,
+ even if it looks like an integer ("1").
+ """
+ data = {
+ "debug": "1", # Special case: must remain string
+ "assets_debug": "1", # Normal case: must be int
+ "debug_nested": {"debug": "0"}, # Recursive test
+ }
+ result = self.session_store.str_to_session(data)
+
+ # "debug" must be string "1", NOT integer 1
+ self.assertEqual(result["debug"], "1")
+ self.assertIsInstance(result["debug"], str)
+ # Other keys do convert
+ self.assertEqual(result["assets_debug"], 1)
+ self.assertIsInstance(result["assets_debug"], int)
+ # Verify exception applies recursively if the key is "debug"
+ self.assertEqual(result["debug_nested"]["debug"], "0")
+ self.assertIsInstance(result["debug_nested"]["debug"], str)
+
+ def test_recursive_traversal(self):
+ """Tests that conversion works in nested structures (lists and dicts)."""
+ data = {"list_of_data": [b"binary_in_list", "100", {"deep_key": "50.5"}]}
+ # 1. Serialize (Bytes -> Str)
+ serialized = self.session_store.session_to_str(data)
+ self.assertTrue(serialized["list_of_data"][0].startswith("base64::"))
+ # 2. Deserialize (Str -> Bytes/Int/Float)
+ # Simulate that JSON already loaded the string "100" and "50.5"
+ # Note: session_to_str does not convert ints to strings, json.dumps does that later.
+ # Here we test str_to_session with data appearing to come from JSON.
+ input_for_read = {
+ "list_of_data": [
+ serialized["list_of_data"][0], # The base64 string
+ "100",
+ {"deep_key": "50.5"},
+ ]
+ }
+ result = self.session_store.str_to_session(input_for_read)
+
+ self.assertEqual(result["list_of_data"][0], b"binary_in_list")
+ self.assertEqual(result["list_of_data"][1], 100)
+ self.assertEqual(result["list_of_data"][2]["deep_key"], 50.5)
+
+ def test_invalid_base64_fallback(self):
+ """If a string has the base64:: prefix but the content is invalid,
+ it must return the original value without crashing.
+ """
+ invalid_data = {"bad_binary": "base64::TESTS"}
+ result = self.session_store.str_to_session(invalid_data)
+ # Should return the original string intact
+ self.assertEqual(result["bad_binary"], "base64::TESTS")