feat(api): filter /rest/v1/standards by saved resource selection (#586)

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()

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
 

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):

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