Backend TCF Purpose Override Support

.fides/fides.toml

@@ -65,3 +65,4 @@ notification_service_type = "mailgun"
 [consent]
 tcf_enabled = false
 ac_enabled = false
+override_vendor_purposes = false

src/fides/api/alembic/migrations/versions/5225ea4de265_tcf_publisher_overrides.py

@@ -0,0 +1,55 @@
+"""tcf_publisher_overrides
+
+Revision ID: 5225ea4de265
+Revises: 1af6950f4625
+Create Date: 2023-11-27 15:35:07.679747
+
+"""
+import sqlalchemy as sa
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "5225ea4de265"
+down_revision = "1af6950f4625"
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    op.create_table(
+        "tcf_publisher_overrides",
+        sa.Column("id", sa.String(length=255), nullable=False),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=True,
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=True,
+        ),
+        sa.Column("purpose", sa.Integer(), nullable=False),
+        sa.Column("is_included", sa.Boolean(), server_default="t", nullable=True),
+        sa.Column("required_legal_basis", sa.String(), nullable=True),
+        sa.PrimaryKeyConstraint("id"),
+    )
+    op.create_index(
+        op.f("ix_tcf_publisher_overrides_id"),
+        "tcf_publisher_overrides",
+        ["id"],
+        unique=False,
+    )
+
+    op.create_unique_constraint(
+        "purpose_constraint", "tcf_publisher_overrides", ["purpose"]
+    )
+
+
+def downgrade():
+    op.drop_index(
+        op.f("ix_tcf_publisher_overrides_id"), table_name="tcf_publisher_overrides"
+    )
+    op.drop_table("tcf_publisher_overrides")

src/fides/api/api/v1/endpoints/system.py

@@ -1,4 +1,4 @@
-from typing import Dict, List, Optional
+from typing import Dict, List, Optional, Union
 
 from fastapi import Depends, HTTPException, Response, Security
 from fastapi_pagination import Page, Params
@@ -258,6 +258,10 @@
     updated_system, _ = await update_system(
         resource, db, current_user.id if current_user else None
     )
+    await supplement_privacy_declaration_response_with_legal_basis_override(
+        updated_system
+    )
+
     return updated_system
 
 
@@ -353,7 +357,11 @@
     Override `System` create/POST to handle `.privacy_declarations` defined inline,
     for backward compatibility and ease of use for API users.
     """
-    return await create_system(resource, db, current_user.id if current_user else None)
+    created = await create_system(
+        resource, db, current_user.id if current_user else None
+    )
+    await supplement_privacy_declaration_response_with_legal_basis_override(created)
+    return created
 
 
 @SYSTEM_ROUTER.get(
@@ -371,7 +379,10 @@
     db: AsyncSession = Depends(get_async_db),
 ) -> List:
     """Get a list of all of the resources of this type."""
-    return await list_resource(System, db)
+    systems = await list_resource(System, db)
+    for system in systems:
+        await supplement_privacy_declaration_response_with_legal_basis_override(system)
+    return systems
 
 
 @SYSTEM_ROUTER.get(
@@ -389,7 +400,10 @@
     db: AsyncSession = Depends(get_async_db),
 ) -> Dict:
     """Get a resource by its fides_key."""
-    return await get_resource_with_custom_fields(System, fides_key, db)
+
+    resp = await get_resource_with_custom_fields(System, fides_key, db)
+    await supplement_privacy_declaration_response_with_legal_basis_override(resp)
+    return resp
 
 
 @SYSTEM_CONNECTION_INSTANTIATE_ROUTER.post(
@@ -412,3 +426,16 @@
 
     system = get_system(db, fides_key)
     return instantiate_connection(db, saas_connector_type, template_values, system)
+
+
+async def supplement_privacy_declaration_response_with_legal_basis_override(resp: Union[Dict, System]) -> None:
+    """At runtime, adds a "legal_basis_for_processing_override" to each PrivacyDeclaration"""
+
+    for privacy_declaration in (
+        resp.get("privacy_declarations")
+        if isinstance(resp, Dict)
+        else resp.privacy_declarations
+    ):
+        privacy_declaration.legal_basis_for_processing_override = (
+            await privacy_declaration.overridden_legal_basis_for_processing
+        )

src/fides/api/app_setup.py

@@ -41,6 +41,7 @@
 from fides.api.service.saas_request.override_implementations import *
 from fides.api.util.cache import get_cache
 from fides.api.util.consent_util import (
+    create_default_tcf_publisher_overrides_on_startup,
     create_tcf_experiences_on_startup,
     load_default_experience_configs_on_startup,
     load_default_notices_on_startup,
@@ -204,7 +205,9 @@
         # Default notices subject to change, so preventing these from
         # loading in test mode to avoid interfering with unit tests.
         load_default_privacy_notices()
+        # Similarly avoiding loading other consent out-of-the-box resources to avoid interfering with unit tests
         load_tcf_experiences()
+        load_tcf_publisher_overrides()
 
     db.close()
 
@@ -255,3 +258,15 @@
         logger.error("Skipping loading TCF Overlay Experiences: {}", str(e))
     finally:
         db.close()
+
+
+def load_tcf_publisher_overrides() -> None:
+    """Load default tcf publisher overrides"""
+    logger.info("Loading default TCF Publisher Overrides")
+    try:
+        db = get_api_session()
+        create_default_tcf_publisher_overrides_on_startup(db)
+    except Exception as e:
+        logger.error("Skipping loading TCF Publisher Overrides: {}", str(e))
+    finally:
+        db.close()

src/fides/api/db/base.py

@@ -39,3 +39,4 @@
 from fides.api.models.system_compass_sync import SystemCompassSync
 from fides.api.models.system_history import SystemHistory
 from fides.api.models.system_manager import SystemManager
+from fides.api.models.tcf_publisher_overrides import TCFPublisherOverride

src/fides/api/models/sql_models.py

@@ -1,14 +1,16 @@
 # type: ignore
-
+# pylint: disable=comparison-with-callable,no-member
 """
 Contains all of the SqlAlchemy models for the Fides resources.
 """
 
 from __future__ import annotations
 
 from enum import Enum as EnumType
-from typing import Any, Dict, List, Optional, Set, Type, TypeVar
+from typing import Any, Dict, List, Optional, Set, Type, TypeVar, Union
 
+from fideslang import MAPPED_PURPOSES_BY_DATA_USE
+from fideslang.gvl import MAPPED_PURPOSES, MappedPurpose
 from fideslang.models import DataCategory as FideslangDataCategory
 from fideslang.models import Dataset as FideslangDataset
 from pydantic import BaseModel
@@ -22,13 +24,21 @@
     Text,
     TypeDecorator,
     UniqueConstraint,
+    case,
     cast,
+    select,
     type_coerce,
 )
 from sqlalchemy.dialects.postgresql import ARRAY, BIGINT, BYTEA
+from sqlalchemy.engine import Row
 from sqlalchemy.exc import IntegrityError
-from sqlalchemy.orm import Session, relationship
-from sqlalchemy.sql import func
+from sqlalchemy.ext.asyncio import AsyncSession, async_object_session
+from sqlalchemy.ext.hybrid import hybrid_property
+from sqlalchemy.orm import Session, object_session, relationship
+from sqlalchemy.orm.attributes import InstrumentedAttribute
+from sqlalchemy.sql import func, Select
+from sqlalchemy.sql.elements import Case
+from sqlalchemy.sql.selectable import ScalarSelect
 from sqlalchemy.sql.sqltypes import DateTime
 from typing_extensions import Protocol, runtime_checkable
 
@@ -38,10 +48,18 @@
 from fides.api.models.client import ClientDetail
 from fides.api.models.fides_user import FidesUser
 from fides.api.models.fides_user_permissions import FidesUserPermissions
+from fides.api.models.tcf_publisher_overrides import TCFPublisherOverride
 from fides.config import get_config
 
 CONFIG = get_config()
 
+# Mapping of data uses to *Purposes* not Special Purposes
+MAPPED_PURPOSES_ONLY_BY_DATA_USE: Dict[str, MappedPurpose] = {
+    data_use: purpose
+    for data_use, purpose in MAPPED_PURPOSES_BY_DATA_USE.items()
+    if purpose in MAPPED_PURPOSES.values()
+}
+
 
 class FidesBase(FideslibBase):
     """
@@ -514,6 +532,124 @@ def create(
         """Overrides base create to avoid unique check on `name` column"""
         return super().create(db=db, data=data, check_name=check_name)
 
+    @hybrid_property
+    def purpose(self) -> Optional[int]:
+        """Returns the instance-level TCF Purpose."""
+        mapped_purpose: Optional[MappedPurpose] = MAPPED_PURPOSES_ONLY_BY_DATA_USE.get(
+            self.data_use
+        )
+        return mapped_purpose.id if mapped_purpose else None
+
+    @purpose.expression
+    def purpose(cls) -> Case:
+        """Returns the class-level TCF Purpose"""
+        return case(
+            [
+                (cls.data_use == data_use, purpose.id)
+                for data_use, purpose in MAPPED_PURPOSES_ONLY_BY_DATA_USE.items()
+            ],
+            else_=None,
+        )
+
+    @hybrid_property
+    async def _publisher_override_legal_basis_join(self) -> Optional[str]:
+        """Returns the instance-level overridden required legal basis"""
+        query: Select = select([TCFPublisherOverride.required_legal_basis]).where(
+            TCFPublisherOverride.purpose == self.purpose
+        )
+        async_session: AsyncSession = async_object_session(self)
+        async with async_session.begin():
+            result = await async_session.execute(query)
+            return result.scalars().first()
+
+    @_publisher_override_legal_basis_join.expression
+    def _publisher_override_legal_basis_join(cls) -> ScalarSelect:
+        """Returns the class-level overridden required legal basis"""
+        return (
+            select([TCFPublisherOverride.required_legal_basis])
+            .where(TCFPublisherOverride.purpose == cls.purpose)
+            .as_scalar()
+        )
+
+    @hybrid_property
+    async def _publisher_override_is_included_join(self) -> Optional[bool]:
+        """Returns the instance-level indication of whether the purpose should be included"""
+        query: Select = select([TCFPublisherOverride.is_included]).where(
+            TCFPublisherOverride.purpose == self.purpose
+        )
+        async_session: AsyncSession = async_object_session(self)
+        async with async_session.begin():
+            result = await async_session.execute(query)
+            return result.scalars().first()
+
+    @_publisher_override_is_included_join.expression
+    def _publisher_override_is_included_join(cls) -> ScalarSelect:
+        """Returns the class-level indication of whether the purpose should be included"""
+        return (
+            select([TCFPublisherOverride.is_included])
+            .where(TCFPublisherOverride.purpose == cls.purpose)
+            .as_scalar()
+        )
+
+    @hybrid_property
+    async def overridden_legal_basis_for_processing(self) -> Optional[str]:
+        """
+        Instance-level override of the legal basis for processing based on
+        publisher preferences.
+        """
+        if not (
+            CONFIG.consent.override_vendor_purposes
+            and self.flexible_legal_basis_for_processing
+        ):
+            return self.legal_basis_for_processing
+
+        is_included: Optional[bool] = await self._publisher_override_is_included_join
+
+        if is_included is False:
+            # Overriding to False to match behavior of class-level override.
+            # Class-level override of legal basis to None removes Privacy Declaration
+            # from Experience
+            return None
+
+        overridden_legal_basis: Optional[str] = (
+            await self._publisher_override_legal_basis_join
+        )
+
+        return (
+            overridden_legal_basis
+            if overridden_legal_basis  # pylint: disable=using-constant-test
+            else self.legal_basis_for_processing
+        )
+
+    @overridden_legal_basis_for_processing.expression
+    def overridden_legal_basis_for_processing(
+        cls,
+    ) -> Union[InstrumentedAttribute, Case]:
+        """
+        Class-level override of the legal basis for processing based on
+        publisher preferences.
+        """
+        if not CONFIG.consent.override_vendor_purposes:
+            return cls.legal_basis_for_processing
+
+        return case(
+            [
+                (
+                    cls.flexible_legal_basis_for_processing.is_(False),
+                    cls.legal_basis_for_processing,
+                ),
+                (
+                    cls._publisher_override_is_included_join.is_(False),
+                    None,
+                ),
+                (
+                    cls._publisher_override_legal_basis_join.is_(None),
+                    cls.legal_basis_for_processing,
+                ),
+            ],
+            else_=cls._publisher_override_legal_basis_join,
+        )
+
 
 class SystemModel(BaseModel):
     fides_key: str

src/fides/api/models/tcf_publisher_overrides.py

@@ -0,0 +1,23 @@
+from sqlalchemy import Boolean, Column, Integer, String, UniqueConstraint
+from sqlalchemy.ext.declarative import declared_attr
+
+from fides.api.db.base_class import Base
+
+
+class TCFPublisherOverride(Base):
+    """
+    Stores TCF Publisher Overrides
+
+    Allows a customer to override Fides-wide which Purposes show up in the TCF Experience, and
+    specify a global legal basis for that Purpose.
+    """
+
+    @declared_attr
+    def __tablename__(self) -> str:
+        return "tcf_publisher_overrides"
+
+    purpose = Column(Integer, nullable=False)
+    is_included = Column(Boolean, server_default="t", default=True)
+    required_legal_basis = Column(String)
+
+    UniqueConstraint("purpose")

src/fides/api/schemas/system.py

@@ -19,6 +19,10 @@ class PrivacyDeclarationResponse(PrivacyDeclaration):
     )
     cookies: Optional[List[Cookies]] = []
 
+    legal_basis_for_processing_override: Optional[str] = Field(
+        description="Global overrides for this purpose's legal basis for processing if applicable.  Defaults to the legal_basis_for_processing otherwise."
+    )
+
 
 class BasicSystemResponse(System):
     """