feature: check auth on sensors referenced in flex-context

Signed-off-by: F.N. Claessen <felix@seita.nl>

flexmeasures/conftest.py

@@ -9,6 +9,7 @@
 import pandas as pd
 import numpy as np
 from flask import request, jsonify
+from flask_login import login_user, logout_user
 from flask_sqlalchemy import SQLAlchemy
 from flask_security import roles_accepted
 from pytest_mock import MockerFixture
@@ -22,6 +23,7 @@
     Gone,
 )
 
+from flexmeasures.api.tests.utils import UserContext
 from flexmeasures.app import create as create_app
 from flexmeasures.auth.policy import ADMIN_ROLE, ADMIN_READER_ROLE
 from flexmeasures.data.services.users import create_user
@@ -1362,3 +1364,37 @@ def add_beliefs(
 @pytest.fixture
 def mock_get_status(mocker: MockerFixture):
     return mocker.patch("flexmeasures.data.services.sensors.get_status", autospec=True)
+
+
+@pytest.fixture
+def requesting_user(request):
+    """Use this fixture to log in a user for the scope of a test.
+
+    Sets the user by passing it an email address (see usage examples below), or pass None to get the AnonymousUser.
+    Passes the user object to the test.
+    Logs the user out after the test ran.
+
+    Usage:
+
+    >>> @pytest.mark.parametrize("requesting_user", ["test_prosumer_user_2@seita.nl", None], indirect=True)
+
+    Or in combination with other parameters:
+
+    @pytest.mark.parametrize(
+        "requesting_user, status_code",
+        [
+            (None, 401),
+            ("test_prosumer_user_2@seita.nl", 200),
+        ],
+        indirect=["requesting_user"],
+    )
+
+    """
+    email = request.param
+    if email is not None:
+        with UserContext(email) as user:
+            login_user(user)
+            yield user
+            logout_user()
+    else:
+        yield

flexmeasures/data/schemas/scheduling/__init__.py

@@ -1,5 +1,8 @@
-from marshmallow import Schema, fields, validate
+from marshmallow import Schema, fields, validate, validates_schema, ValidationError
+from werkzeug.exceptions import Forbidden
 
+from flexmeasures.auth.policy import check_access
+from flexmeasures.data.schemas.scheduling.utils import find_sensors
 from flexmeasures.data.schemas.sensors import QuantityOrSensor, SensorIdField
 
 
@@ -29,3 +32,15 @@ class FlexContextSchema(Schema):
     inflexible_device_sensors = fields.List(
         SensorIdField(), data_key="inflexible-device-sensors"
     )
+
+    @validates_schema
+    def check_read_access_on_sensors(self, data: dict, **kwargs):
+        sensors = find_sensors(data)
+        for sensor, field_name in sensors:
+            try:
+                check_access(context=sensor, permission="read")
+            except Forbidden:
+                raise ValidationError(
+                    message=f"User has no read access to sensor {sensor.id}.",
+                    field_name=self.fields[field_name].data_key,
+                )

flexmeasures/data/schemas/scheduling/utils.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from flexmeasures import Sensor
+
+
+def find_sensors(data, parent_key='', path='') -> list[tuple[Sensor, str]]:
+    """Recursively find all sensors in a nested dictionary or list along with the fields referring to them."""
+    sensors = []
+
+    if isinstance(data, dict):
+        for key, value in data.items():
+            new_parent_key = f"{parent_key}.{key}" if parent_key else key
+            new_path = f"{path}.{key}" if path else key
+            if isinstance(value, Sensor):
+                sensors.append((value, f"{new_parent_key}{path}"))
+            else:
+                sensors.extend(find_sensors(value, new_parent_key, new_path))
+    elif isinstance(data, list):
+        for index, item in enumerate(data):
+            new_parent_key = f"{parent_key}[{index}]"
+            new_path = f"{path}[{index}]"
+            sensors.extend(find_sensors(item, new_parent_key, new_path))
+
+    return sensors

flexmeasures/data/schemas/tests/conftest.py

@@ -1,25 +1,80 @@
 import pytest
 from datetime import timedelta
 
-from flexmeasures.data.models.time_series import Sensor, TimedBelief
+from flask_security import SQLAlchemySessionUserDatastore, hash_password
+
+from flexmeasures import Sensor, User, UserRole
+from flexmeasures.data.models.time_series import TimedBelief
+from flexmeasures.data.models.user import Account
 from flexmeasures.data.models.generic_assets import GenericAsset, GenericAssetType
 
 
 @pytest.fixture(scope="module")
-def dummy_asset(db, app):
+def dummy_accounts(db, app):
+    dummy_account_1 = Account(name="dummy account 1")
+    db.session.add(dummy_account_1)
+
+    dummy_account_2 = Account(name="dummy account 2")
+    db.session.add(dummy_account_2)
+
+    # Assign account IDs
+    db.session.flush()
+
+    return {
+        dummy_account_1.name: dummy_account_1,
+        dummy_account_2.name: dummy_account_2,
+    }
+
+
+@pytest.fixture(scope="module")
+def dummy_user(db, app, dummy_accounts):
+    user_datastore = SQLAlchemySessionUserDatastore(db.session, User, UserRole)
+    user = user_datastore.create_user(
+        username="dummy user",
+        email="dummy_user@seita.nl",
+        password=hash_password("testtest"),
+        account_id=dummy_accounts["dummy account 1"].id,
+        active=True,
+    )
+    return user
+
+
+@pytest.fixture(scope="module")
+def dummy_assets(db, app, dummy_accounts):
     dummy_asset_type = GenericAssetType(name="DummyGenericAssetType")
     db.session.add(dummy_asset_type)
 
-    _dummy_asset = GenericAsset(
-        name="DummyGenericAsset", generic_asset_type=dummy_asset_type
+    dummy_asset_1 = GenericAsset(
+        name="dummy asset 1",
+        generic_asset_type=dummy_asset_type,
+        owner=dummy_accounts["dummy account 1"],
+    )
+    db.session.add(dummy_asset_1)
+
+    dummy_asset_2 = GenericAsset(
+        name="dummy asset 2",
+        generic_asset_type=dummy_asset_type,
+        owner=dummy_accounts["dummy account 1"],
+    )
+    db.session.add(dummy_asset_2)
+
+    dummy_asset_3 = GenericAsset(
+        name="dummy asset 3",
+        generic_asset_type=dummy_asset_type,
+        owner=dummy_accounts["dummy account 2"],
     )
-    db.session.add(_dummy_asset)
+    db.session.add(dummy_asset_3)
 
-    return _dummy_asset
+    return {
+        dummy_asset_1.name: dummy_asset_1,
+        dummy_asset_2.name: dummy_asset_2,
+        dummy_asset_3.name: dummy_asset_3,
+    }
 
 
 @pytest.fixture(scope="module")
-def setup_dummy_sensors(db, app, dummy_asset):
+def setup_dummy_sensors(db, app, dummy_assets):
+    dummy_asset = dummy_assets["dummy asset 1"]
     sensor1 = Sensor(
         "sensor 1",
         generic_asset=dummy_asset,
@@ -58,7 +113,8 @@ def setup_dummy_sensors(db, app, dummy_asset):
 
 
 @pytest.fixture(scope="module")
-def setup_efficiency_sensors(db, app, dummy_asset):
+def setup_efficiency_sensors(db, app, dummy_assets):
+    dummy_asset = dummy_assets["dummy asset 1"]
     sensor = Sensor(
         "efficiency",
         generic_asset=dummy_asset,
@@ -72,7 +128,8 @@ def setup_efficiency_sensors(db, app, dummy_asset):
 
 
 @pytest.fixture(scope="module")
-def setup_site_capacity_sensor(db, app, dummy_asset, setup_sources):
+def setup_site_capacity_sensor(db, app, dummy_assets, setup_sources):
+    dummy_asset = dummy_assets["dummy asset 1"]
     sensor = Sensor(
         "site-power-capacity",
         generic_asset=dummy_asset,

flexmeasures/data/schemas/tests/test_scheduling.py

@@ -239,7 +239,10 @@ def load_schema():
         ),
     ],
 )
-def test_flex_context_schema(db, app, setup_site_capacity_sensor, flex_context, fails):
+@pytest.mark.parametrize(
+    "requesting_user", ["dummy_user@seita.nl"], indirect=True
+)
+def test_flex_context_schema(db, app, setup_site_capacity_sensor, flex_context, fails, dummy_user, requesting_user):
     schema = FlexContextSchema()
 
     # Replace sensor name with sensor ID