diff --git a/application/tests/resource_filter_test.py b/application/tests/resource_filter_test.py new file mode 100644 index 000000000..99995d0c2 --- /dev/null +++ b/application/tests/resource_filter_test.py @@ -0,0 +1,149 @@ +"""Tests for server-side resource-selection filtering on /rest/v1/standards. + +Part of #586 (PR3). A logged-in user with a saved selection sees only their +selected standards (plus OPENCRE); everyone else sees the full list. +``Node_collection.standards`` (Neo4j) is mocked to a fixed list; the User and +selection rows are real on Postgres. +""" + +import json +import os +import unittest +from typing import Any, Dict, List +from unittest.mock import patch + +from application import create_app, sqla +from application.database import db +from application.utils.gap_analysis import OPENCRE_STANDARD_NAME + +STANDARDS = ["ASVS", "CWE", "SAMM", "ZAP"] +FULL = sorted(STANDARDS + [OPENCRE_STANDARD_NAME]) + + +class TestStandardsFilter(unittest.TestCase): + def setUp(self) -> None: + self._prev_no_load_graph = os.environ.get("NO_LOAD_GRAPH_DB") + os.environ["NO_LOAD_GRAPH_DB"] = "1" + self.app = create_app(mode="test") + self.app.secret_key = "test-secret" + self.app_context = self.app.app_context() + self.app_context.push() + sqla.create_all() + self.collection = db.Node_collection() + + def tearDown(self) -> None: + sqla.session.remove() + sqla.drop_all() + self.app_context.pop() + if self._prev_no_load_graph is None: + os.environ.pop("NO_LOAD_GRAPH_DB", None) + else: + os.environ["NO_LOAD_GRAPH_DB"] = self._prev_no_load_graph + + def _login(self, client: Any, google_sub: str = "sub-1", name: str = "U") -> None: + with client.session_transaction() as sess: + sess["google_id"] = google_sub + sess["name"] = name + + def _seed_selection(self, names: List[str]) -> None: + user = self.collection.upsert_user( + google_sub="sub-1", email="a@x.com", display_name="U" + ) + if names: + self.collection.set_user_resource_selection(user.id, names) + + @staticmethod + def _enabled() -> Dict[str, str]: + return { + "CRE_ENABLE_LOGIN": "1", + "CRE_ENABLE_MYOPENCRE": "1", + "INSECURE_REQUESTS": "1", + } + + # --- the filter applies --- + @patch.object(db.Node_collection, "standards") + def test_logged_in_with_selection_filters(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection(["ASVS", "CWE"]) + with patch.dict(os.environ, self._enabled()): + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards") + self.assertEqual(resp.status_code, 200) + # Only the selection + OPENCRE; SAMM and ZAP are dropped. + self.assertEqual( + sorted(json.loads(resp.data)), + sorted(["ASVS", "CWE", OPENCRE_STANDARD_NAME]), + ) + + @patch.object(db.Node_collection, "standards") + def test_opencre_always_kept_even_if_not_selected( + self, standards_mock: Any + ) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection(["ASVS"]) # OPENCRE not in the selection + with patch.dict(os.environ, self._enabled()): + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards") + body = json.loads(resp.data) + self.assertIn(OPENCRE_STANDARD_NAME, body) + self.assertEqual(sorted(body), sorted(["ASVS", OPENCRE_STANDARD_NAME])) + + # --- no-op conditions: full list --- + @patch.object(db.Node_collection, "standards") + def test_logged_in_empty_selection_returns_full(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection([]) # user exists, no selection + with patch.dict(os.environ, self._enabled()): + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards") + self.assertEqual(sorted(json.loads(resp.data)), FULL) + + @patch.object(db.Node_collection, "standards") + def test_anonymous_returns_full(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + with patch.dict(os.environ, self._enabled()): + with self.app.test_client() as client: + resp = client.get("/rest/v1/standards") + self.assertEqual(sorted(json.loads(resp.data)), FULL) + + @patch.object(db.Node_collection, "standards") + def test_login_off_returns_full(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection(["ASVS", "CWE"]) + env = {"CRE_ENABLE_MYOPENCRE": "1", "INSECURE_REQUESTS": "1"} + with patch.dict(os.environ, env): + os.environ.pop("CRE_ENABLE_LOGIN", None) + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards") + self.assertEqual(sorted(json.loads(resp.data)), FULL) + + @patch.object(db.Node_collection, "standards") + def test_myopencre_off_returns_full(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection(["ASVS", "CWE"]) + env = {"CRE_ENABLE_LOGIN": "1", "INSECURE_REQUESTS": "1"} + with patch.dict(os.environ, env): + os.environ.pop("CRE_ENABLE_MYOPENCRE", None) + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards") + self.assertEqual(sorted(json.loads(resp.data)), FULL) + + # --- ?all=true bypass --- + @patch.object(db.Node_collection, "standards") + def test_all_true_bypasses_filter(self, standards_mock: Any) -> None: + standards_mock.return_value = list(STANDARDS) + self._seed_selection(["ASVS", "CWE"]) + with patch.dict(os.environ, self._enabled()): + with self.app.test_client() as client: + self._login(client) + resp = client.get("/rest/v1/standards?all=true") + self.assertEqual(sorted(json.loads(resp.data)), FULL) + + +if __name__ == "__main__": + unittest.main() diff --git a/application/web/openapi_registry.py b/application/web/openapi_registry.py index 438111f91..7a65b0239 100644 --- a/application/web/openapi_registry.py +++ b/application/web/openapi_registry.py @@ -59,6 +59,7 @@ class PathSpec: "extra_responses", "response_override", "request_body", + "parameters", ) def __init__( @@ -76,6 +77,7 @@ def __init__( extra_responses: Optional[Dict[str, Any]] = None, response_override: Optional[Dict[str, Any]] = None, request_body: Optional[Dict[str, Any]] = None, + parameters: Optional[List[Dict[str, Any]]] = None, ) -> None: self.path = path self.method = method.lower() @@ -89,6 +91,7 @@ def __init__( self.extra_responses = extra_responses or {} self.response_override = response_override self.request_body = request_body + self.parameters = parameters OPENAPI_PATHS: List[PathSpec] = [ @@ -201,6 +204,24 @@ def __init__( "standards", tags=["Standards"], summary="List standards", + description=( + "For a logged-in user with a saved resource selection (and the " + "MyOpenCRE feature enabled), the list is restricted to that selection " + "(OpenCRE is always included). Pass all=true to bypass the filter and " + "return every standard." + ), + parameters=[ + { + "name": "all", + "in": "query", + "required": False, + "schema": {"type": "boolean"}, + "description": ( + "When true, return all standards even if the user has a saved " + "selection." + ), + } + ], not_found=False, response_override={ "200": { @@ -494,6 +515,10 @@ def _operation_from_path( parameters.extend( [param for param in query_params if param["name"] not in path_names] ) + if path_spec.parameters: + parameters.extend( + [param for param in path_spec.parameters if param["name"] not in path_names] + ) if parameters: operation["parameters"] = parameters diff --git a/application/web/web_main.py b/application/web/web_main.py index dbac65eb1..4075eda61 100644 --- a/application/web/web_main.py +++ b/application/web/web_main.py @@ -542,6 +542,7 @@ def standards() -> Any: standards = list(database.standards()) if OPENCRE_STANDARD_NAME not in standards: standards.append(OPENCRE_STANDARD_NAME) + standards = apply_user_resource_filter(database, standards) return standards @@ -938,6 +939,29 @@ def _resolve_current_user(database): return user +def apply_user_resource_filter(database, names): + """Restrict ``names`` to the current user's saved resource selection. + + No-op (returns ``names`` unchanged) unless login AND MyOpenCRE are enabled, a + user is resolved, and that user has a non-empty selection — same discipline + as the resource-selection API. ``?all=true`` bypasses the filter for a single + request. ``OPENCRE_STANDARD_NAME`` is always kept (it is the core graph and is + already special-cased). Part of #586. + """ + if request.args.get("all") == "true": + return names + if not (is_login_enabled() and is_myopencre_enabled()): + return names + user = _resolve_current_user(database) + if user is None: + return names + selection = database.get_user_resource_selection(user.id) + if not selection: + return names + keep = set(selection) | {OPENCRE_STANDARD_NAME} + return [name for name in names if name in keep] + + def admin_imports_enabled_required(f): @wraps(f) def enabled_r(*args, **kwargs): diff --git a/docs/api/openapi.yaml b/docs/api/openapi.yaml index 4bf657765..3708aa0f4 100644 --- a/docs/api/openapi.yaml +++ b/docs/api/openapi.yaml @@ -415,6 +415,17 @@ paths: tags: - Standards summary: List standards + description: For a logged-in user with a saved resource selection (and the MyOpenCRE + feature enabled), the list is restricted to that selection (OpenCRE is always + included). Pass all=true to bypass the filter and return every standard. + parameters: + - name: all + in: query + required: false + schema: + type: boolean + description: When true, return all standards even if the user has a saved + selection. responses: '200': description: Standards retrieved