Add option to unescape SafeStr (#3144)

Co-authored-by: Thomas <[email protected]>
Co-authored-by: Thomas <[email protected]>

CHANGELOG.md

@@ -33,13 +33,15 @@ The types of changes are:
 ### Changed
 
 - Merge instances of RTK `createApi` into one instance for better cache invalidation [#3059](https://github.com/ethyca/fides/pull/3059)
+- Explicitly escape/unescape certain fields instead of using SafeStr [#3144](https://github.com/ethyca/fides/pull/3144)
 - Update custom field definition uniqueness to be case insensitive name per resource type [#3215](https://github.com/ethyca/fides/pull/3215)
 - Restrict where privacy notices of certain consent mechanisms must be displayed [#3195](https://github.com/ethyca/fides/pull/3195)
 - Merged the `lib` submodule into the `api.ops` submodule [#3134](https://github.com/ethyca/fides/pull/3134)
 - Merged duplicate privacy declaration components [#3254](https://github.com/ethyca/fides/pull/3254)
 - Refactor client applications into a monorepo with turborepo, extract fides-js into a standalone package, and improve privacy-center to load configuration at runtime [#3105](https://github.com/ethyca/fides/pull/3105)
 
 ### Fixed
+
 - Prevent ability to unintentionally show "default" Privacy Center configuration, styles, etc. [#3242](https://github.com/ethyca/fides/pull/3242)
 - Fix broken links to docs site pages in Admin UI [#3232](https://github.com/ethyca/fides/pull/3232)
 - Repoint legacy docs site links to the new and improved docs site [#3167](https://github.com/ethyca/fides/pull/3167)

clients/admin-ui/src/features/common/CommonHeaders.ts

@@ -7,5 +7,6 @@ export function addCommonHeaders(headers: Headers, token: string | null) {
   if (token) {
     headers.set("authorization", `Bearer ${token}`);
   }
+  headers.set("Unescape-Safestr", "true");
   return headers;
 }

src/fides/api/ops/api/v1/endpoints/privacy_notice_endpoints.py

@@ -1,6 +1,7 @@
+from html import escape, unescape
 from typing import Dict, List, Optional, Tuple
 
-from fastapi import Depends, Security
+from fastapi import Depends, Request, Security
 from fastapi_pagination import Page, Params
 from fastapi_pagination.bases import AbstractPage
 from fastapi_pagination.ext.sqlalchemy import paginate
@@ -18,6 +19,7 @@
 from fides.api.ops.api import deps
 from fides.api.ops.api.v1 import scope_registry
 from fides.api.ops.api.v1 import urn_registry as urls
+from fides.api.ops.api.v1.endpoints.utils import transform_fields
 from fides.api.ops.common_exceptions import ValidationError
 from fides.api.ops.models.privacy_notice import (
     PrivacyNotice,
@@ -31,6 +33,7 @@
 router = APIRouter(tags=["Privacy Notice"], prefix=urls.V1_URL_PREFIX)
 
 DataUseMap = Dict[str, List[schemas.PrivacyNoticeResponse]]
+ESCAPE_FIELDS = ["name", "description", "internal_description", "origin"]
 
 
 def generate_notice_query(
@@ -68,6 +71,7 @@ def get_privacy_notice_list(
     show_disabled: Optional[bool] = True,
     region: Optional[PrivacyNoticeRegion] = None,
     systems_applicable: Optional[bool] = False,
+    request: Request,
 ) -> AbstractPage[PrivacyNotice]:
     """
     Return a paginated list of `PrivacyNotice` records in this system.
@@ -80,8 +84,15 @@ def get_privacy_notice_list(
         systems_applicable=systems_applicable,
         region=region,
     )
+    should_unescape = request.headers.get("unescape-safestr")
     privacy_notices = notice_query.order_by(PrivacyNotice.created_at.desc())
-    return paginate(privacy_notices, params=params)
+    paginated = paginate(privacy_notices, params=params)
+    if should_unescape:
+        paginated.items = [  # type: ignore[attr-defined]
+            transform_fields(transformation=unescape, model=item, fields=ESCAPE_FIELDS)
+            for item in paginated.items  # type: ignore[attr-defined]
+        ]
+    return paginated
 
 
 @router.get(
@@ -161,11 +172,20 @@ def get_privacy_notice(
     *,
     privacy_notice_id: str,
     db: Session = Depends(deps.get_db),
+    request: Request,
 ) -> PrivacyNotice:
     """
     Return a single PrivacyNotice
     """
-    return get_privacy_notice_or_error(db, privacy_notice_id)  # type: ignore[return-value]
+    should_unescape = request.headers.get("unescape-safestr")
+    notice = get_privacy_notice_or_error(db, privacy_notice_id)
+    if should_unescape:
+        notice = transform_fields(
+            transformation=unescape,
+            model=notice,
+            fields=ESCAPE_FIELDS,
+        )  # type: ignore
+    return notice
 
 
 def validate_notice_data_uses(
@@ -218,12 +238,22 @@ def create_privacy_notices(
     except ValidationError as e:
         raise HTTPException(HTTP_422_UNPROCESSABLE_ENTITY, detail=e.message)
 
-    return [
-        PrivacyNotice.create(
-            db=db, data=privacy_notice.dict(exclude_unset=True), check_name=False
+    # Loop through and create the new privacy notices
+    created_privacy_notices: List[PrivacyNotice] = []
+    for privacy_notice in privacy_notices:
+        privacy_notice = transform_fields(
+            transformation=escape,
+            model=privacy_notice,
+            fields=ESCAPE_FIELDS,
         )
-        for privacy_notice in privacy_notices
-    ]
+        created_privacy_notice = PrivacyNotice.create(
+            db=db,
+            data=privacy_notice.dict(exclude_unset=True),
+            check_name=False,
+        )
+        created_privacy_notices.append(created_privacy_notice)
+
+    return created_privacy_notices
 
 
 def prepare_privacy_notice_patches(
@@ -253,6 +283,11 @@ def prepare_privacy_notice_patches(
                 detail=f"No PrivacyNotice found for id {update_data.id}.",
             )
 
+        update_data = transform_fields(
+            transformation=escape,
+            model=update_data,
+            fields=ESCAPE_FIELDS,
+        )  # type: ignore
         updates_and_existing.append((update_data, existing_notices[update_data.id]))
 
     # we temporarily store proposed update data in-memory for validation purposes only

src/fides/api/ops/api/v1/endpoints/utils.py

@@ -1,7 +1,7 @@
 from __future__ import annotations
 
 from datetime import datetime
-from typing import List, Optional, Tuple
+from typing import Callable, List, Optional, Tuple
 
 from fastapi import HTTPException
 from starlette.status import HTTP_400_BAD_REQUEST
@@ -28,3 +28,17 @@ def validate_start_and_end_filters(
                 status_code=HTTP_400_BAD_REQUEST,
                 detail=f"Value specified for {field_name}_lt: {end} must be after {field_name}_gt: {start}.",
             )
+
+
+def transform_fields(
+    transformation: Callable, model: object, fields: List[str]
+) -> object:
+    """
+    Takes a callable and returns a transformed object.
+    """
+
+    for name, value in {field: getattr(model, field) for field in fields}.items():
+        if value:
+            setattr(model, name, transformation(value))
+
+    return model

src/fides/api/ops/schemas/privacy_notice.py

@@ -5,7 +5,6 @@
 
 from pydantic import Extra, conlist, root_validator, validator
 
-from fides.api.custom_types import SafeStr
 from fides.api.ops.models.privacy_notice import (
     ConsentMechanism,
     EnforcementLevel,
@@ -22,13 +21,13 @@ class PrivacyNotice(FidesSchema):
     stricter but not less strict
     """
 
-    name: Optional[SafeStr]
-    description: Optional[SafeStr]
-    internal_description: Optional[SafeStr]
-    origin: Optional[SafeStr]
+    name: Optional[str]
+    description: Optional[str]
+    internal_description: Optional[str]
+    origin: Optional[str]
     regions: Optional[conlist(PrivacyNoticeRegion, min_items=1)]  # type: ignore
     consent_mechanism: Optional[ConsentMechanism]
-    data_uses: Optional[conlist(SafeStr, min_items=1)]  # type: ignore
+    data_uses: Optional[conlist(str, min_items=1)]  # type: ignore
     enforcement_level: Optional[EnforcementLevel]
     disabled: Optional[bool] = False
     has_gpc_flag: Optional[bool] = False
@@ -104,10 +103,10 @@ class PrivacyNoticeCreation(PrivacyNotice):
     It also establishes some fields _required_ for creation
     """
 
-    name: SafeStr
+    name: str
     regions: conlist(PrivacyNoticeRegion, min_items=1)  # type: ignore
     consent_mechanism: ConsentMechanism
-    data_uses: conlist(SafeStr, min_items=1)  # type: ignore
+    data_uses: conlist(str, min_items=1)  # type: ignore
     enforcement_level: EnforcementLevel
 
 

tests/ops/api/v1/endpoints/test_privacy_notice_endpoints.py

@@ -579,6 +579,56 @@ def test_pagination(
             most_recent = privacy_notices.pop()
             assert privacy_notice["name"] == most_recent.name
 
+    @pytest.mark.usefixtures(
+        "system",
+    )
+    def test_can_unescape(
+        self,
+        db,
+        api_client: TestClient,
+        generate_auth_header,
+        url,
+    ):
+        auth_header = generate_auth_header(
+            scopes=[scopes.PRIVACY_NOTICE_READ, scopes.PRIVACY_NOTICE_CREATE]
+        )
+
+        escaped_name = "Data Sales 'n' stuff"
+        unescaped_name = "Data Sales 'n' stuff"
+        PrivacyNotice.create(
+            db=db,
+            data={
+                "name": escaped_name,
+                "description": "a sample privacy notice configuration",
+                "regions": [PrivacyNoticeRegion.us_ca],
+                "consent_mechanism": ConsentMechanism.opt_in,
+                "data_uses": ["advertising"],
+                "enforcement_level": EnforcementLevel.system_wide,
+                "displayed_in_overlay": True,
+            },
+        )
+
+        # without the unescaped header, should return the escaped version
+        resp = api_client.get(
+            url,
+            headers=auth_header,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        print(data["items"])
+        assert data["items"][0]["name"] == escaped_name
+
+        # now request with the unescape header
+        unescape_header = {"Unescape-Safestr": "yes"}
+        auth_and_unescape_header = {**auth_header, **unescape_header}
+        resp = api_client.get(
+            url,
+            headers=auth_and_unescape_header,
+        )
+        data = resp.json()
+        print(data["items"])
+        assert data["items"][0]["name"] == unescaped_name
+
 
 class TestGetPrivacyNoticeDetail:
     @pytest.fixture(scope="function")
@@ -649,6 +699,67 @@ def test_get_privacy_notice(
         assert data["disabled"] == privacy_notice.disabled
         assert data["displayed_in_overlay"] == privacy_notice.displayed_in_overlay
 
+    def test_get_privacy_notice_unescaped(
+        self,
+        api_client: TestClient,
+        generate_auth_header,
+    ):
+        auth_header = generate_auth_header(
+            scopes=[scopes.PRIVACY_NOTICE_READ, scopes.PRIVACY_NOTICE_CREATE]
+        )
+
+        # post a new privacy notice that has some sneaky characters
+        maybe_dangerous_description = "user's description <script />"
+        resp = api_client.post(
+            V1_URL_PREFIX + PRIVACY_NOTICE,
+            headers=auth_header,
+            json=[
+                {
+                    "name": "test privacy notice 1",
+                    "description": maybe_dangerous_description,
+                    "origin": "privacy_notice_template_1",
+                    "regions": [
+                        PrivacyNoticeRegion.eu_be.value,
+                        PrivacyNoticeRegion.us_ca.value,
+                    ],
+                    "consent_mechanism": ConsentMechanism.opt_in.value,
+                    "data_uses": ["advertising"],
+                    "enforcement_level": EnforcementLevel.system_wide.value,
+                    "displayed_in_overlay": True,
+                }
+            ],
+        )
+        print(f"Created Notice: {resp.text}")
+        assert resp.status_code == 200
+        created_notice = resp.json()[0]
+
+        url = V1_URL_PREFIX + PRIVACY_NOTICE_DETAIL.format(
+            privacy_notice_id=created_notice["id"]
+        )
+
+        # without the unescaped header, should return the escaped version
+        resp = api_client.get(
+            url,
+            # headers={**auth_header, **unescape_header},
+            headers=auth_header,
+        )
+        assert resp.status_code == 200
+        data = resp.json()
+        print(f"Unescaped Response: {data}")
+        assert data["description"] == "user&#x27;s description &lt;script /&gt;"
+
+        # now request with the unescape header
+        unescape_header = {"Unescape-Safestr": "yes"}
+        auth_and_unescape_header = {**auth_header, **unescape_header}
+        print(f"Auth & Unescape Headers: {auth_and_unescape_header}")
+        resp = api_client.get(
+            url,
+            headers=auth_and_unescape_header,
+        )
+        data = resp.json()
+        print(f"Escaped Data: {data}")
+        assert data["description"] == maybe_dangerous_description
+
 
 class TestGetPrivacyNoticesByDataUse:
     @pytest.fixture(scope="function")

tests/ops/api/v1/test_utils.py

@@ -1,9 +1,14 @@
 from datetime import datetime
+from html import escape
 
 import pytest
 from fastapi import HTTPException
 
-from fides.api.ops.api.v1.endpoints.utils import validate_start_and_end_filters
+from fides.api.ops.api.v1.endpoints.utils import (
+    transform_fields,
+    validate_start_and_end_filters,
+)
+from fides.api.ops.schemas.privacy_notice import PrivacyNotice
 
 
 class TestValidateStartAndEndFilters:
@@ -25,3 +30,15 @@ def test_valid_dates(self):
         validate_start_and_end_filters(
             [(datetime(2020, 2, 2), datetime(2020, 2, 1), "created")]
         )
+
+
+class TestTransformFields:
+    def test_transform_escape(self):
+        escaped_field = "user's description <script />"
+        notice_1 = PrivacyNotice(name="whew", description=escaped_field)
+
+        unescaped_field = "user's description <script />"
+        notice_2 = PrivacyNotice(name="whew", description=unescaped_field)
+        notice_2 = transform_fields(escape, notice_2, ["description"])
+
+        assert notice_1 == notice_2