associate users with stewarded monitors; encode monitors in jwt token

Author: adamsachs

src/fides/api/cryptography/schemas/jwt.py

@@ -4,3 +4,4 @@
 JWE_PAYLOAD_ROLES = "roles"
 JWE_PAYLOAD_SYSTEMS = "systems"
 JWE_PAYLOAD_CONNECTIONS = "connections"
+JWE_PAYLOAD_MONITORS = "monitors"

src/fides/api/models/client.py

@@ -17,6 +17,7 @@
     JWE_ISSUED_AT,
     JWE_PAYLOAD_CLIENT_ID,
     JWE_PAYLOAD_CONNECTIONS,
+    JWE_PAYLOAD_MONITORS,
     JWE_PAYLOAD_ROLES,
     JWE_PAYLOAD_SCOPES,
     JWE_PAYLOAD_SYSTEMS,
@@ -30,6 +31,7 @@
 DEFAULT_ROLES: list[str] = []
 DEFAULT_SYSTEMS: list[str] = []
 DEFAULT_CONNECTIONS: list[str] = []
+DEFAULT_MONITORS: list[str] = []
 
 
 class ClientDetail(Base):
@@ -47,6 +49,7 @@ def __tablename__(self) -> str:
     connections = Column(
         ARRAY(String), nullable=False, server_default="{}", default=dict
     )
+    monitors = Column(ARRAY(String), nullable=False, server_default="{}", default=dict)
     fides_key = Column(String, index=True, unique=True, nullable=True)
     user_id = Column(
         String, ForeignKey(FidesUser.id_field_path), nullable=True, unique=True
@@ -67,6 +70,7 @@ def create_client_and_secret(
         roles: list[str] | None = None,
         systems: list[str] | None = None,
         connections: list[str] | None = None,
+        monitors: list[str] | None = None,
         in_memory: bool | None = False,
     ) -> tuple["ClientDetail", str]:
         """Creates a ClientDetail and returns that along with the unhashed secret
@@ -88,6 +92,9 @@ def create_client_and_secret(
         if not connections:
             connections = DEFAULT_CONNECTIONS
 
+        if not monitors:
+            monitors = DEFAULT_MONITORS
+
         salt = generate_salt()
         hashed_secret = hash_credential_with_salt(
             secret.encode(encoding),
@@ -104,6 +111,7 @@ def create_client_and_secret(
             "roles": roles,
             "systems": systems,
             "connections": connections,
+            "monitors": monitors,
         }
 
         if in_memory:
@@ -142,6 +150,7 @@ def create_access_code_jwe(self, encryption_key: str) -> str:
             JWE_PAYLOAD_ROLES: self.roles,
             JWE_PAYLOAD_SYSTEMS: self.systems,
             JWE_PAYLOAD_CONNECTIONS: self.connections,
+            JWE_PAYLOAD_MONITORS: self.monitors,
         }
         return generate_jwe(json.dumps(payload), encryption_key)
 
@@ -176,6 +185,7 @@ def _get_root_client_detail(
             roles=roles,
             systems=[],
             connections=[],
+            monitors=[],
         )
 
     return ClientDetail(
@@ -186,4 +196,5 @@ def _get_root_client_detail(
         roles=DEFAULT_ROLES,
         systems=DEFAULT_SYSTEMS,
         connections=DEFAULT_CONNECTIONS,
+        monitors=DEFAULT_MONITORS,
     )

src/fides/api/models/detection_discovery/core.py

@@ -213,6 +213,7 @@ class MonitorConfig(Base):
     stewards = relationship(
         FidesUser,
         secondary="monitorsteward",
+        back_populates="stewarded_monitors",
         lazy="selectin",
     )
 

src/fides/api/models/fides_user.py

@@ -26,6 +26,7 @@
 from fides.config import CONFIG
 
 if TYPE_CHECKING:
+    from fides.api.models.detection_discovery import MonitorConfig
     from fides.api.models.fides_user_permissions import FidesUserPermissions
     from fides.api.models.fides_user_respondent_email_verification import (
         FidesUserRespondentEmailVerification,
@@ -77,6 +78,15 @@ class FidesUser(Base):
     systems = relationship(
         "System", secondary="systemmanager", back_populates="data_stewards"
     )  # type: ignore
+
+    # Monitors for which this user is a steward
+    stewarded_monitors = relationship(
+        "MonitorConfig",
+        secondary="monitorsteward",
+        back_populates="stewards",
+        lazy="selectin",
+    )  # type: ignore
+
     # permissions relationship is defined via backref in FidesUserPermissions
     email_verifications = relationship(
         "FidesUserRespondentEmailVerification",
@@ -104,6 +114,11 @@ class FidesUser(Base):
     def system_ids(self) -> List[str]:
         return [system.id for system in self.systems]
 
+    @property
+    def stewarded_monitor_ids(self) -> List[str]:
+        """Returns list of monitor IDs for which this user is a steward."""
+        return [monitor.id for monitor in self.stewarded_monitors]
+
     @classmethod
     def hash_password(cls, password: str, encoding: str = "UTF-8") -> tuple[str, str]:
         """Utility function to hash a user's password with a generated salt"""