La 202 update testing for flushing oauth tokens (#5587)

src/fides/api/schemas/saas/strategy_configuration.py

@@ -1,7 +1,7 @@
 from enum import Enum
 from typing import Any, Dict, List, Optional, Union
 
-from pydantic import BaseModel, ConfigDict, field_validator, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
 
 from fides.api.schemas.saas.saas_config import Header, QueryParam, SaaSRequest
 from fides.api.schemas.saas.shared_schemas import (
@@ -154,8 +154,19 @@ class OAuth2BaseConfiguration(StrategyConfiguration):
 
 class OAuth2AuthorizationCodeConfiguration(OAuth2BaseConfiguration):
     """
+    Oauth Authorization that requires manual user interaction to get authorization
     The standard OAuth2 configuration but with an additional property to configure
     the authorization request for the Authorization Code flow.
     """
 
     authorization_request: SaaSRequest
+
+
+class OAuth2ClientCredentialsConfiguration(OAuth2BaseConfiguration):
+    """
+    Ouath authorization that does not require manual user interation to get authorization
+    The standard OAuth2 configuration, but excluding the refresh token during logging
+    since the client credentials flow does not require a refresh token.
+    """
+
+    refresh_request: Optional[SaaSRequest] = Field(exclude=True)

tests/fixtures/saas_example_fixtures.py

@@ -19,6 +19,7 @@
 from fides.api.schemas.saas.saas_config import ParamValue
 from fides.api.schemas.saas.strategy_configuration import (
     OAuth2AuthorizationCodeConfiguration,
+    OAuth2ClientCredentialsConfiguration,
 )
 from fides.api.service.masking.strategy.masking_strategy_nullify import (
     NullMaskingStrategy,
@@ -390,6 +391,75 @@ def oauth2_authorization_code_connection_config(
     connection_config.delete(db)
 
 
+## TODO: base on the previous connection config to set up a new improved
+
+
+@pytest.fixture(scope="function")
+def oauth2_client_credentials_configuration() -> OAuth2ClientCredentialsConfiguration:
+    return {
+        "token_request": {
+            "method": "POST",
+            "path": "/oauth/token",
+            "headers": [
+                {
+                    "name": "Content-Type",
+                    "value": "application/x-www-form-urlencoded",
+                }
+            ],
+            "query_params": [
+                {"name": "client_id", "value": "<client_id>"},
+                {"name": "client_secret", "value": "<client_secret>"},
+                {"name": "grant_type", "value": "client_credentials"},
+            ],
+        },
+    }
+
+
+@pytest.fixture(scope="function")
+def oauth2_client_credentials_connection_config(
+    db: Session, oauth2_client_credentials_configuration
+) -> Generator:
+    secrets = {
+        "domain": "localhost",
+        "client_id": "client",
+        "client_secret": "secret",
+        "access_token": "access",
+    }
+    saas_config = {
+        "fides_key": "oauth2_client_credentials_connector",
+        "name": "OAuth2 Client Credentials Connector",
+        "type": "custom",
+        "description": "Generic OAuth2 connector for testing",
+        "version": "0.0.1",
+        "connector_params": [{"name": item} for item in secrets.keys()],
+        "client_config": {
+            "protocol": "https",
+            "host": secrets["domain"],
+            "authentication": {
+                "strategy": "oauth2_client_credentials",
+                "configuration": oauth2_client_credentials_configuration,
+            },
+        },
+        "endpoints": [],
+        "test_request": {"method": "GET", "path": "/test"},
+    }
+
+    fides_key = saas_config["fides_key"]
+    connection_config = ConnectionConfig.create(
+        db=db,
+        data={
+            "key": fides_key,
+            "name": fides_key,
+            "connection_type": ConnectionType.saas,
+            "access": AccessLevel.write,
+            "secrets": secrets,
+            "saas_config": saas_config,
+        },
+    )
+    yield connection_config
+    connection_config.delete(db)
+
+
 @pytest.fixture(scope="session")
 def saas_config() -> Dict[str, Any]:
     saas_config = {}

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

@@ -112,13 +112,21 @@ def connections():
 
 class TestPatchSystemConnections:
     @pytest.fixture(scope="function")
-    def system_linked_with_connection_config(
+    def system_linked_with_oauth2_authorization_code_connection_config(
         self, system: System, oauth2_authorization_code_connection_config, db: Session
     ):
         system.connection_configs = oauth2_authorization_code_connection_config
         db.commit()
         return system
 
+    @pytest.fixture(scope="function")
+    def system_linked_with_oauth2_client_credentials_connection_config(
+        self, system: System, oauth2_client_credentials_connection_config, db: Session
+    ):
+        system.connection_configs = oauth2_client_credentials_connection_config
+        db.commit()
+        return system
+
     def test_patch_connections_valid_system(
         self, api_client: TestClient, generate_auth_header, url, payload
     ):
@@ -205,12 +213,42 @@ def test_patch_connections_role_check_viewer(
         resp = api_client.patch(url, headers=auth_header, json=payload)
         assert resp.status_code == expected_status_code
 
-    def test_patch_connection_secrets_removes_access_token(
+    def test_patch_connection_secrets_removes_access_token_for_clients_credentials(
+        self,
+        api_client: TestClient,
+        generate_auth_header,
+        url,
+        system_linked_with_oauth2_client_credentials_connection_config,
+    ):
+        auth_header = generate_auth_header(
+            scopes=[CONNECTION_READ, CONNECTION_CREATE_OR_UPDATE]
+        )
+
+        # verify the connection_config is authorized
+        resp = api_client.get(url, headers=auth_header)
+
+        assert resp.status_code == HTTP_200_OK
+        assert resp.json()["items"][0]["authorized"] is True
+
+        # patch the connection_config with new secrets (but no access_token)
+        resp = api_client.patch(
+            f"{url}/secrets?verify=False",
+            headers=auth_header,
+            json={"domain": "test_domain"},
+        )
+
+        # verify the connection_config is no longer authorized
+        resp = api_client.get(url, headers=auth_header)
+
+        assert resp.status_code == HTTP_200_OK
+        assert resp.json()["items"][0]["authorized"] is False
+
+    def test_patch_connection_secrets_removes_access_token_for_client_config(
         self,
         api_client: TestClient,
         generate_auth_header,
         url,
-        system_linked_with_connection_config,
+        system_linked_with_oauth2_authorization_code_connection_config,
     ):
         auth_header = generate_auth_header(
             scopes=[CONNECTION_READ, CONNECTION_CREATE_OR_UPDATE]
@@ -429,7 +467,7 @@ def url(self, system) -> str:
         return V1_URL_PREFIX + f"/system/{system.fides_key}/connection"
 
     @pytest.fixture(scope="function")
-    def system_linked_with_connection_config(
+    def system_linked_with_oauth2_authorization_code_connection_config(
         self, system: System, connection_config, db: Session
     ):
         system.connection_configs = connection_config
@@ -465,11 +503,13 @@ def test_delete_connection_config(
         api_client: TestClient,
         db: Session,
         generate_auth_header,
-        system_linked_with_connection_config,
+        system_linked_with_oauth2_authorization_code_connection_config,
     ) -> None:
         auth_header = generate_auth_header(scopes=[CONNECTION_DELETE])
         # the key needs to be cached before the delete
-        key = system_linked_with_connection_config.connection_configs.key
+        key = (
+            system_linked_with_oauth2_authorization_code_connection_config.connection_configs.key
+        )
         resp = api_client.delete(url, headers=auth_header)
         assert resp.status_code == HTTP_204_NO_CONTENT
         assert db.query(ConnectionConfig).filter_by(key=key).first() is None
@@ -576,13 +616,13 @@ def test_delete_connection_configs_role_viewer(
         acting_user_role,
         expected_status_code,
         assign_system,
-        system_linked_with_connection_config,
+        system_linked_with_oauth2_authorization_code_connection_config,
         request,
         db: Session,
     ) -> None:
         url = (
             V1_URL_PREFIX
-            + f"/system/{system_linked_with_connection_config.fides_key}/connection"
+            + f"/system/{system_linked_with_oauth2_authorization_code_connection_config.fides_key}/connection"
         )
 
         acting_user_role = request.getfixturevalue(acting_user_role)
@@ -595,10 +635,12 @@ def test_delete_connection_configs_role_viewer(
             api_client.put(
                 assign_url,
                 headers=system_manager_auth_header,
-                json=[system_linked_with_connection_config.fides_key],
+                json=[
+                    system_linked_with_oauth2_authorization_code_connection_config.fides_key
+                ],
             )
             auth_header = generate_system_manager_header(
-                [system_linked_with_connection_config.id]
+                [system_linked_with_oauth2_authorization_code_connection_config.id]
             )
         else:
             auth_header = generate_role_header_for_user(
@@ -622,13 +664,13 @@ def test_delete_connection_configs_role_check(
         generate_auth_header,
         acting_user_role,
         expected_status_code,
-        system_linked_with_connection_config,
+        system_linked_with_oauth2_authorization_code_connection_config,
         request,
         db: Session,
     ) -> None:
         url = (
             V1_URL_PREFIX
-            + f"/system/{system_linked_with_connection_config.fides_key}/connection"
+            + f"/system/{system_linked_with_oauth2_authorization_code_connection_config.fides_key}/connection"
         )
 
         acting_user_role = request.getfixturevalue(acting_user_role)

tests/ops/service/authentication/test_authentication_strategy_oauth2_client_credentials.py

@@ -8,11 +8,6 @@
 from sqlalchemy.orm import Session
 
 from fides.api.common_exceptions import FidesopsException, OAuth2TokenException
-from fides.api.models.connectionconfig import (
-    AccessLevel,
-    ConnectionConfig,
-    ConnectionType,
-)
 from fides.api.service.authentication.authentication_strategy import (
     AuthenticationStrategy,
 )
@@ -21,74 +16,6 @@
 )
 
 
-@pytest.fixture(scope="function")
-def oauth2_client_credentials_configuration() -> (
-    OAuth2ClientCredentialsAuthenticationStrategy
-):
-    return {
-        "token_request": {
-            "method": "POST",
-            "path": "/oauth/token",
-            "headers": [
-                {
-                    "name": "Content-Type",
-                    "value": "application/x-www-form-urlencoded",
-                }
-            ],
-            "query_params": [
-                {"name": "client_id", "value": "<client_id>"},
-                {"name": "client_secret", "value": "<client_secret>"},
-                {"name": "grant_type", "value": "client_credentials"},
-            ],
-        }
-    }
-
-
-@pytest.fixture(scope="function")
-def oauth2_client_credentials_connection_config(
-    db: Session, oauth2_client_credentials_configuration
-) -> Generator:
-    secrets = {
-        "domain": "localhost",
-        "client_id": "client",
-        "client_secret": "secret",
-        "access_token": "access",
-    }
-    saas_config = {
-        "fides_key": "oauth2_client_credentials_connector",
-        "name": "OAuth2 Client Credentials Connector",
-        "type": "custom",
-        "description": "Generic OAuth2 connector for testing",
-        "version": "0.0.1",
-        "connector_params": [{"name": item} for item in secrets.keys()],
-        "client_config": {
-            "protocol": "https",
-            "host": secrets["domain"],
-            "authentication": {
-                "strategy": "oauth2_client_credentials",
-                "configuration": oauth2_client_credentials_configuration,
-            },
-        },
-        "endpoints": [],
-        "test_request": {"method": "GET", "path": "/test"},
-    }
-
-    fides_key = saas_config["fides_key"]
-    connection_config = ConnectionConfig.create(
-        db=db,
-        data={
-            "key": fides_key,
-            "name": fides_key,
-            "connection_type": ConnectionType.saas,
-            "access": AccessLevel.write,
-            "secrets": secrets,
-            "saas_config": saas_config,
-        },
-    )
-    yield connection_config
-    connection_config.delete(db)
-
-
 class TestAddAuthentication:
     # happy path, being able to use the existing access token
     def test_oauth2_authentication(