You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@allura.apache.org by cc...@apache.org on 2026/08/21 04:25:00 UTC
(allura) 01/01: [#8617] Update User's set_tool_data to store encrypted tool data values
This is an automated email from the ASF dual-hosted git repository.
asf-gitbox-commits pushed a commit to branch cc/8617-part1
in repository https://gitbox.apache.org/repos/asf/allura.git
commit c40e41d424341ec86ac82a2425bf22094c8d018d
Author: Carlos Cruz <ca...@slashdotmedia.com>
AuthorDate: Thu Aug 20 22:24:36 2026 -0600
[#8617] Update User's set_tool_data to store encrypted tool data values
---
Allura/allura/model/auth.py | 7 ++-
Allura/allura/tests/model/test_auth.py | 29 +++++++++++++
.../tests/scripts/test_convert_encrypted_field.py | 27 ++++++++++++
scripts/convert_encrypted_field.py | 50 ++++++++++++++++++----
4 files changed, 104 insertions(+), 9 deletions(-)
diff --git a/Allura/allura/model/auth.py b/Allura/allura/model/auth.py
index 4b9c34094..cdc1ba90e 100644
--- a/Allura/allura/model/auth.py
+++ b/Allura/allura/model/auth.py
@@ -792,9 +792,14 @@ def get_tool_data(self, tool, key, default=None):
else:
return result
- def set_tool_data(self, tool, **kw):
+ def set_tool_data(self, tool, *, encrypt=False, **kw):
d = self.tool_data.setdefault(tool, {})
d.update(kw)
+ if encrypt:
+ d.update({
+ f'{key}_encrypted': self.encr(value)
+ for key, value in kw.items()
+ })
state(self).soil()
def address_object(self, addr):
diff --git a/Allura/allura/tests/model/test_auth.py b/Allura/allura/tests/model/test_auth.py
index d5d2161c7..e11f7418d 100644
--- a/Allura/allura/tests/model/test_auth.py
+++ b/Allura/allura/tests/model/test_auth.py
@@ -494,6 +494,35 @@ def test_preference_email_address_updates_encrypted_field(self):
assert raw_preferences['email_address_encrypted'] == M.User.encr('new@example.com')
assert M.User.decr(raw_preferences['email_address_encrypted']) == 'new@example.com'
+ def test_set_tool_data_can_dual_write_encrypted_values(self):
+ user = M.User(
+ username='tool-data-encrypted-setter-test',
+ tool_data={'sfx': {'unrelated': 'keep'}},
+ )
+ user.set_tool_data(
+ 'sfx',
+ encrypt=True,
+ registration_ip='198.51.100.12',
+ registration_rdns=None,
+ normalized_email='',
+ )
+ user.set_tool_data('temp_reg_fields', encrypt=True, phone_ext='104')
+ user.set_tool_data('sfx', registration_ua='TestBrowser/1.0')
+ ThreadLocalODMSession.flush_all()
+
+ tool_data = state(user).document['tool_data']
+ assert tool_data['sfx']['registration_ip'] == '198.51.100.12'
+ assert tool_data['sfx']['registration_ip_encrypted'] == M.User.encr('198.51.100.12')
+ assert tool_data['sfx']['registration_rdns'] is None
+ assert tool_data['sfx']['registration_rdns_encrypted'] is None
+ assert tool_data['sfx']['normalized_email'] == ''
+ assert tool_data['sfx']['normalized_email_encrypted'] == M.User.encr('')
+ assert tool_data['temp_reg_fields']['phone_ext'] == '104'
+ assert tool_data['temp_reg_fields']['phone_ext_encrypted'] == M.User.encr('104')
+ assert tool_data['sfx']['registration_ua'] == 'TestBrowser/1.0'
+ assert 'registration_ua_encrypted' not in tool_data['sfx']
+ assert tool_data['sfx']['unrelated'] == 'keep'
+
def test_personal_data_fields_are_stored_encrypted_on_creation(self):
user = M.User(
username='personal-data-encrypted-create-test',
diff --git a/Allura/allura/tests/scripts/test_convert_encrypted_field.py b/Allura/allura/tests/scripts/test_convert_encrypted_field.py
index e5415cc38..1af6d4e88 100644
--- a/Allura/allura/tests/scripts/test_convert_encrypted_field.py
+++ b/Allura/allura/tests/scripts/test_convert_encrypted_field.py
@@ -15,6 +15,8 @@
# specific language governing permissions and limitations
# under the License.
+import pytest
+
from ming import schema as S
from ming.encryption import NestedEncryptedProperty
from ming.odm import FieldProperty
@@ -37,6 +39,10 @@ def encr(cls, value):
return None if value is None else f'encrypted:{value}'.encode()
+class DynamicToolDataModel:
+ tool_data = FieldProperty({str: {str: None}})
+
+
def test_encryption_schema_info_uses_encrypted_nested_array_path():
field_schema, traverses_array = _encryption_schema_info(
TestModel,
@@ -65,6 +71,27 @@ class PreMigrationTestModel:
assert traverses_array is True
+def test_encryption_schema_info_supports_opted_in_dynamic_field():
+ field_schema, traverses_array = _encryption_schema_info(
+ DynamicToolDataModel,
+ 'tool_data.sfx.registration_ip',
+ 'tool_data.sfx.registration_ip_encrypted',
+ encrypt_dynamic_field=True,
+ )
+
+ assert isinstance(field_schema, S.Binary)
+ assert traverses_array is False
+
+
+def test_encryption_schema_info_rejects_dynamic_field_without_opt_in():
+ with pytest.raises(AssertionError, match='--encrypt-dynamic-field'):
+ _encryption_schema_info(
+ DynamicToolDataModel,
+ 'tool_data.sfx.registration_ip',
+ 'tool_data.sfx.registration_ip_encrypted',
+ )
+
+
def test_encrypt_nested_array_value_updates_every_item():
socialnetworks = [
{'socialnetwork': 'Mastodon', 'accounturl': 'https://example.com/@user'},
diff --git a/scripts/convert_encrypted_field.py b/scripts/convert_encrypted_field.py
index aba21ea42..c02499895 100644
--- a/scripts/convert_encrypted_field.py
+++ b/scripts/convert_encrypted_field.py
@@ -45,7 +45,7 @@ def _split_field_path(field_name: str) -> list[str]:
return field_path
-def _schema_info_for_field_path(Model: type[MappedClass], field_name: str):
+def _schema_path_info(Model: type[MappedClass], field_name: str):
field_path = _split_field_path(field_name)
top_level_name = field_path[0]
try:
@@ -57,6 +57,7 @@ def _schema_info_for_field_path(Model: type[MappedClass], field_name: str):
current_schema = top_level_prop.field.schema
traverses_array = False
+ traverses_dynamic_object = False
for i, path_part in enumerate(field_path[1:], start=1):
traversed = '.'.join(field_path[:i])
while isinstance(current_schema, schema.Array):
@@ -65,25 +66,55 @@ def _schema_info_for_field_path(Model: type[MappedClass], field_name: str):
if not isinstance(current_schema, schema.Object):
raise AssertionError(
f'Invalid nested field path {field_name!r}; {traversed!r} is not an object field')
- if path_part not in current_schema.fields:
+ if path_part in current_schema.fields:
+ current_schema = current_schema.fields[path_part]
+ continue
+
+ if len(current_schema.fields) == 1:
+ key_schema, field_schema = next(iter(current_schema.fields.items()))
+ else:
+ key_schema = None
+ if isinstance(key_schema, str) or key_schema is None:
raise MissingFieldPathError(
f'Invalid nested field path {field_name!r}; missing key {traversed + "." + path_part!r}')
- current_schema = current_schema.fields[path_part]
+ try:
+ schema.SchemaItem.make(key_schema).validate(path_part)
+ except schema.Invalid as e:
+ raise MissingFieldPathError(
+ f'Invalid nested field path {field_name!r}; missing key '
+ f'{traversed + "." + path_part!r}') from e
+ current_schema = field_schema
+ traverses_dynamic_object = True
+
+ return current_schema, traverses_array, traverses_dynamic_object
+
- return current_schema, traverses_array
+def _schema_info_for_field_path(Model: type[MappedClass], field_name: str):
+ field_schema, traverses_array, _ = _schema_path_info(Model, field_name)
+ return field_schema, traverses_array
def _encryption_schema_info(Model: type[MappedClass], plain_field_name: str,
- encrypted_field_name: str):
+ encrypted_field_name: str, *, encrypt_dynamic_field=False):
try:
- return _schema_info_for_field_path(Model, encrypted_field_name)
+ field_schema, traverses_array, traverses_dynamic_object = _schema_path_info(
+ Model, encrypted_field_name)
except MissingFieldPathError:
# Pre-migration support: infer array traversal from the plaintext
# schema when the encrypted field has not been added to the model yet.
_, traverses_array = _schema_info_for_field_path(Model, plain_field_name)
return None, traverses_array
+ if traverses_dynamic_object and not isinstance(field_schema, schema.Binary):
+ if not encrypt_dynamic_field:
+ raise AssertionError(
+ f'{encrypted_field_name!r} uses a dynamic schema; '
+ 'pass --encrypt-dynamic-field to convert it')
+ field_schema = schema.Binary()
+
+ return field_schema, traverses_array
+
def _get_nested_value(rec: dict, field_name: str):
value = rec
@@ -203,12 +234,14 @@ def _update_nested_array_records(raw_collection, plain_field_name, transform, li
def main(class_name: str, plain_field_name: str,
- *, remove_unencrypted: bool = False, redo_all: bool = False, limit: int | None = None):
+ *, remove_unencrypted: bool = False, redo_all: bool = False,
+ encrypt_dynamic_field: bool = False, limit: int | None = None):
"""
:param class_name: full class name, e.g. allura.model.user.User
:param plain_field_name: name of the unencrypted field, e.g. display_name
:param remove_unencrypted: WARNING only run this after your codebase is already on the latest code
:param redo_all: re-encrypt records that already have encrypted values (in case they changed since last run)
+ :param encrypt_dynamic_field: treat a field in a dynamic object schema as an encrypted scalar
:param limit: convert this many records per update type (bulk, individual, removal) default All
"""
encrypted_field_name = _default_encrypted_field_name(plain_field_name)
@@ -219,7 +252,8 @@ def main(class_name: str, plain_field_name: str,
# sanity checks that the fields are correct and ready
encr_schema, traverses_array = _encryption_schema_info(
- Model, plain_field_name, encrypted_field_name)
+ Model, plain_field_name, encrypted_field_name,
+ encrypt_dynamic_field=encrypt_dynamic_field)
if encr_schema is not None:
assert isinstance(encr_schema, schema.Binary) or _is_encrypted_list_schema(encr_schema)
elif remove_unencrypted: