From c4812f65a85a63fc26f35769911c18c236a1e3aa Mon Sep 17 00:00:00 2001 From: skypank Date: Sat, 8 Aug 2026 16:07:05 +0530 Subject: [PATCH 1/3] fix(db): concurrency-safe set_user_resource_selection + enforce OpenCRE in selection (Part of #586) --- application/database/db.py | 35 ++++++--- application/tests/user_model_test.py | 45 +++++++++++ application/tests/user_resources_api_test.py | 83 ++++++++++++++++++-- application/web/web_main.py | 5 ++ 4 files changed, 150 insertions(+), 18 deletions(-) diff --git a/application/database/db.py b/application/database/db.py index a7c50b443..a96340cf1 100644 --- a/application/database/db.py +++ b/application/database/db.py @@ -1352,19 +1352,30 @@ def set_user_resource_selection( for name in standard_names: if name not in deduped: deduped.append(name) - self.session.query(UserResourceSelection).filter( - UserResourceSelection.user_id == user_id - ).delete() - for name in deduped: - self.session.add( - UserResourceSelection( - id=generate_uuid(), - user_id=user_id, - standard_name=name, - created_at=now, + + def _replace() -> None: + self.session.query(UserResourceSelection).filter( + UserResourceSelection.user_id == user_id + ).delete() + for name in deduped: + self.session.add( + UserResourceSelection( + id=generate_uuid(), + user_id=user_id, + standard_name=name, + created_at=now, + ) ) - ) - self.session.commit() + self.session.commit() + + try: + _replace() + except IntegrityError: + # A concurrent PUT for the same user committed the same rows between + # our delete and insert; roll back and retry the replace exactly once + # (mirrors upsert_user). A second failure propagates — no retry loop. + self.session.rollback() + _replace() return self.get_user_resource_selection(user_id) def __get_external_links(self) -> List[Tuple[CRE, Node, str]]: diff --git a/application/tests/user_model_test.py b/application/tests/user_model_test.py index bf241de03..e29faa294 100644 --- a/application/tests/user_model_test.py +++ b/application/tests/user_model_test.py @@ -2,6 +2,8 @@ import os import unittest +from typing import Any +from unittest.mock import patch from sqlalchemy.exc import IntegrityError @@ -159,6 +161,49 @@ def test_deleting_user_cascades_to_selection(self) -> None: sqla.session.commit() self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 0) + def test_set_resource_selection_recovers_from_integrity_error(self) -> None: + # A concurrent PUT can make the first commit raise IntegrityError on + # uq_user_resource_selection. The method must roll back and retry once, + # then return the correct selection. + user = self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + real_commit = self.collection.session.commit + calls = {"n": 0} + + def flaky_commit(*args: Any, **kwargs: Any) -> None: + calls["n"] += 1 + if calls["n"] == 1: + raise IntegrityError( + "stmt", {}, Exception("uq_user_resource_selection") + ) + real_commit() + + with patch.object(self.collection.session, "commit", side_effect=flaky_commit): + result = self.collection.set_user_resource_selection( + user.id, ["ASVS", "CWE"] + ) + + self.assertEqual(sorted(result), ["ASVS", "CWE"]) + self.assertEqual(calls["n"], 2) # retried exactly once + + def test_set_resource_selection_reraises_on_persistent_integrity_error( + self, + ) -> None: + # If the retry also fails, the error propagates — the recovery must not + # loop indefinitely (retry exactly once). + user = self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + + def always_raise(*args: Any, **kwargs: Any) -> None: + raise IntegrityError("stmt", {}, Exception("uq_user_resource_selection")) + + with patch.object(self.collection.session, "commit", side_effect=always_raise): + with self.assertRaises(IntegrityError): + self.collection.set_user_resource_selection(user.id, ["ASVS"]) + self.collection.session.rollback() + if __name__ == "__main__": unittest.main() diff --git a/application/tests/user_resources_api_test.py b/application/tests/user_resources_api_test.py index 2d51f865e..f8eb019cf 100644 --- a/application/tests/user_resources_api_test.py +++ b/application/tests/user_resources_api_test.py @@ -12,6 +12,7 @@ from application import create_app, sqla from application.database import db +from application.utils.gap_analysis import OPENCRE_STANDARD_NAME class TestUserResourcesApi(unittest.TestCase): @@ -145,11 +146,13 @@ def test_put_persists_and_returns_selection(self) -> None: ) self.assertEqual(resp.status_code, 200) self.assertEqual( - sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"] + sorted(json.loads(resp.data)["selected"]), + sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]), ) get = client.get("/rest/v1/user/resources") self.assertEqual( - sorted(json.loads(get.data)["selected"]), ["ASVS", "CWE"] + sorted(json.loads(get.data)["selected"]), + sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]), ) def test_put_replaces_previous_selection(self) -> None: @@ -169,7 +172,10 @@ def test_put_replaces_previous_selection(self) -> None: self._login(client, "sub-1", "U") client.put("/rest/v1/user/resources", json={"selected": ["SAMM"]}) get = client.get("/rest/v1/user/resources") - self.assertEqual(json.loads(get.data)["selected"], ["SAMM"]) + self.assertEqual( + sorted(json.loads(get.data)["selected"]), + sorted(["SAMM", OPENCRE_STANDARD_NAME]), + ) def test_put_dedupes_input(self) -> None: self.collection.upsert_user( @@ -190,7 +196,8 @@ def test_put_dedupes_input(self) -> None: json={"selected": ["ASVS", "ASVS", "CWE"]}, ) self.assertEqual( - sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"] + sorted(json.loads(resp.data)["selected"]), + sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]), ) def test_put_trims_and_dedupes_whitespace_variants(self) -> None: @@ -215,9 +222,73 @@ def test_put_trims_and_dedupes_whitespace_variants(self) -> None: ) self.assertEqual(resp.status_code, 200) self.assertEqual( - sorted(json.loads(resp.data)["selected"]), ["ASVS", "CWE"] + sorted(json.loads(resp.data)["selected"]), + sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]), + ) + # ASVS, CWE, and the always-injected OpenCRE. + self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 3) + + def test_put_always_persists_opencre(self) -> None: + # A non-empty selection without OpenCRE gets OpenCRE injected at write. + self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + with patch.dict( + os.environ, + { + "CRE_ENABLE_LOGIN": "1", + "CRE_ENABLE_MYOPENCRE": "1", + "INSECURE_REQUESTS": "1", + }, + ): + with self.app.test_client() as client: + self._login(client, "sub-1", "U") + resp = client.put( + "/rest/v1/user/resources", json={"selected": ["ASVS"]} + ) + self.assertEqual(resp.status_code, 200) + self.assertIn(OPENCRE_STANDARD_NAME, json.loads(resp.data)["selected"]) + + def test_put_opencre_not_duplicated(self) -> None: + self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + with patch.dict( + os.environ, + { + "CRE_ENABLE_LOGIN": "1", + "CRE_ENABLE_MYOPENCRE": "1", + "INSECURE_REQUESTS": "1", + }, + ): + with self.app.test_client() as client: + self._login(client, "sub-1", "U") + resp = client.put( + "/rest/v1/user/resources", + json={"selected": ["ASVS", OPENCRE_STANDARD_NAME]}, ) - self.assertEqual(sqla.session.query(db.UserResourceSelection).count(), 2) + selected = json.loads(resp.data)["selected"] + self.assertEqual(selected.count(OPENCRE_STANDARD_NAME), 1) + + def test_put_empty_selection_stays_empty(self) -> None: + # Empty PUT must remain [] (PR3 treats [] as "show all") — OpenCRE must + # NOT be injected, or that would become "show only OpenCRE". + self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + with patch.dict( + os.environ, + { + "CRE_ENABLE_LOGIN": "1", + "CRE_ENABLE_MYOPENCRE": "1", + "INSECURE_REQUESTS": "1", + }, + ): + with self.app.test_client() as client: + self._login(client, "sub-1", "U") + resp = client.put("/rest/v1/user/resources", json={"selected": []}) + self.assertEqual(resp.status_code, 200) + self.assertEqual(json.loads(resp.data)["selected"], []) def test_put_400_on_invalid_body(self) -> None: self.collection.upsert_user( diff --git a/application/web/web_main.py b/application/web/web_main.py index 4075eda61..f337df42a 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -1376,6 +1376,11 @@ def put_user_resources() -> Any: # Normalize before storing: otherwise " ASVS " and "ASVS" both validate but # persist as distinct rows, defeating the dedupe. selected = [name.strip() for name in raw_selected] + # OpenCRE is always part of a non-empty selection (matches the read filter and + # the "OpenCRE is always included" UI copy). An empty selection stays empty — + # [] means "show everything", so injecting OpenCRE would wrongly narrow it. + if selected and OPENCRE_STANDARD_NAME not in selected: + selected.append(OPENCRE_STANDARD_NAME) database = db.Node_collection() user = _resolve_current_user(database) if user is None: From 95ae2b2694b2fb625cf977aaef8e5cb0f1bff8d2 Mon Sep 17 00:00:00 2001 From: skypank Date: Sat, 8 Aug 2026 17:36:54 +0530 Subject: [PATCH 2/3] fix(db): serialize set_user_resource_selection with a per-user row lock (Part of #586) --- application/database/db.py | 26 ++++++++---- application/tests/user_model_test.py | 61 +++++++++++++++++++++++++++- 2 files changed, 79 insertions(+), 8 deletions(-) diff --git a/application/database/db.py b/application/database/db.py index a96340cf1..db24ea3bb 100644 --- a/application/database/db.py +++ b/application/database/db.py @@ -1353,7 +1353,16 @@ def set_user_resource_selection( if name not in deduped: deduped.append(name) - def _replace() -> None: + def _replace() -> List[str]: + # Serialize concurrent replacements for this user by locking the + # parent User row: a second PUT for the same user waits here until + # ours commits, then its DELETE sees our rows. Without this, two + # *disjoint* selections could both commit (no IntegrityError) and + # merge, breaking replacement semantics. FOR UPDATE is a no-op on + # SQLite (dev/tests). + self.session.query(User).filter( + User.id == user_id + ).with_for_update().first() self.session.query(UserResourceSelection).filter( UserResourceSelection.user_id == user_id ).delete() @@ -1366,17 +1375,20 @@ def _replace() -> None: created_at=now, ) ) + # Capture our selection while still holding the lock, so the return + # value can't reflect a later concurrent write. + stored = self.get_user_resource_selection(user_id) self.session.commit() + return stored try: - _replace() + return _replace() except IntegrityError: - # A concurrent PUT for the same user committed the same rows between - # our delete and insert; roll back and retry the replace exactly once - # (mirrors upsert_user). A second failure propagates — no retry loop. + # Same-key collision on uq_user_resource_selection; roll back and + # retry the replace exactly once (mirrors upsert_user). A second + # failure propagates — no retry loop. self.session.rollback() - _replace() - return self.get_user_resource_selection(user_id) + return _replace() def __get_external_links(self) -> List[Tuple[CRE, Node, str]]: external_links: List[Tuple[CRE, Node, str]] = [] diff --git a/application/tests/user_model_test.py b/application/tests/user_model_test.py index e29faa294..811bc697f 100644 --- a/application/tests/user_model_test.py +++ b/application/tests/user_model_test.py @@ -1,8 +1,9 @@ """Tests for user persistence and per-user resource selection (issue #586, RFC #876 TODO 1/2).""" import os +import threading import unittest -from typing import Any +from typing import Any, List from unittest.mock import patch from sqlalchemy.exc import IntegrityError @@ -204,6 +205,64 @@ def always_raise(*args: Any, **kwargs: Any) -> None: self.collection.set_user_resource_selection(user.id, ["ASVS"]) self.collection.session.rollback() + def test_set_resource_selection_blocks_on_locked_user_row(self) -> None: + # The replacement serializes per user by locking the User row: a second + # writer must WAIT for the first to commit (so its DELETE then sees the + # first's rows, instead of the two selections merging). Reproduced + # deterministically: hold the user-row lock here, then assert a worker's + # set_user_resource_selection blocks until we release it. Postgres-only — + # FOR UPDATE is a no-op on SQLite, so nothing would block there. + if "postgresql" not in str(sqla.engine.url): + self.skipTest("row-lock serialization requires Postgres (FOR UPDATE)") + + user = self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + user_id = user.id + self.collection.set_user_resource_selection(user_id, ["ASVS"]) # seed + self.collection.session.rollback() + + # Acquire and HOLD the user-row lock on this session (open transaction). + self.collection.session.query(db.User).filter( + db.User.id == user_id + ).with_for_update().first() + + done = threading.Event() + errors: List[Exception] = [] + + def worker() -> None: + with self.app.app_context(): + try: + # Empty replacement = DELETE only, no INSERT. So the only + # thing that can make this touch (and block on) the User row + # is the method's own FOR UPDATE — which isolates it from the + # FK lock an INSERT would otherwise take. + db.Node_collection().set_user_resource_selection(user_id, []) + except Exception as e: + errors.append(e) + finally: + sqla.session.remove() + done.set() + + t = threading.Thread(target=worker) + t.start() + + # While we hold the lock the worker must block — it cannot finish. + self.assertFalse( + done.wait(timeout=2), "worker did not block on the user-row lock" + ) + + # Release the lock; the worker now proceeds and completes. + self.collection.session.rollback() + self.assertTrue( + done.wait(timeout=15), "worker did not finish after lock release" + ) + t.join(timeout=5) + + self.assertEqual(errors, []) + self.collection.session.rollback() # fresh snapshot + self.assertEqual(self.collection.get_user_resource_selection(user_id), []) + if __name__ == "__main__": unittest.main() From 9bfb579ee386d64460ca54831c0f65ce515092a7 Mon Sep 17 00:00:00 2001 From: skypank Date: Sat, 8 Aug 2026 17:44:34 +0530 Subject: [PATCH 3/3] fix(db): serialize set_user_resource_selection with a per-user row lock (Part of #586) --- application/tests/user_model_test.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/application/tests/user_model_test.py b/application/tests/user_model_test.py index 811bc697f..5423539a4 100644 --- a/application/tests/user_model_test.py +++ b/application/tests/user_model_test.py @@ -238,7 +238,7 @@ def worker() -> None: # is the method's own FOR UPDATE — which isolates it from the # FK lock an INSERT would otherwise take. db.Node_collection().set_user_resource_selection(user_id, []) - except Exception as e: + except Exception as e: # noqa: BLE001 - report worker failures to test errors.append(e) finally: sqla.session.remove()