Commit ae4bff0c authored by George Papathanail's avatar George Papathanail
Browse files

fix: don't echo notificationAuthToken in the create-session response

parent b1e7d9b2

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+7 −1
Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 7 added lines, 1 removed line.
Original line number Diff line number Diff line
@@ -9,6 +9,7 @@ from open_exposure_gateway.api.camara.quality_on_demand.v0_10_1.schemas import ( 
    CreateSession,

    QosStatus,

    SessionInfo,

    Webhook,

)

from open_exposure_gateway.application.callback_secrets import encrypt_secret, load_bearer_token

from open_exposure_gateway.application.mappers.quality_on_demand_mapper import (
@@ -242,7 +243,12 @@ class QualityOnDemandService: 
            devicePorts=request.devicePorts,

            applicationServerPorts=request.applicationServerPorts,

            qosProfile=request.qosProfile,

            webhook=request.webhook,

            # Never echo the secret notificationAuthToken back.

            webhook=(

                Webhook(notificationUrl=request.webhook.notificationUrl)

                if request.webhook is not None

                else None

            ),

            duration=request.duration,

            startedAt=started_at,

            expiresAt=expires_at,

tests/unit/test_qod_flows.py

+14 −0
Changes for tests/unit/test_qod_flows.py: 14 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -65,6 +65,14 @@ class TestQodSessionFlow: 
        body = {**SESSION_BODY, "webhook": {"notificationUrl": "https://application-server.com"}}

        response = api_client.post(f"{QOD_BASE}/sessions", json=body)

        assert response.status_code == 201

        assert response.json()["webhook"] == {"notificationUrl": "https://application-server.com"}



    def test_create_session_without_webhook_has_no_webhook_key(

        self, api_client: TestClient

    ) -> None:

        response = api_client.post(f"{QOD_BASE}/sessions", json=SESSION_BODY)

        assert response.status_code == 201

        assert "webhook" not in response.json()



    def test_srm_receives_a_valid_activate_command(

        self, api_client: TestClient, fake_bus: FakeDataBus
@@ -371,6 +379,12 @@ class TestNotificationAuthTokenWithKeyFlow: 
    def callback_secret_key(self) -> bytes:

        return bytes(range(32))



    def test_create_response_omits_the_token(self, api_client: TestClient) -> None:

        response = api_client.post(f"{QOD_BASE}/sessions", json=AUTH_WEBHOOK_BODY)

        assert response.status_code == 201

        assert response.json()["webhook"] == {"notificationUrl": "https://client.example.com/cb"}

        assert NOTIFICATION_AUTH_TOKEN not in response.text



    async def test_token_is_sent_as_bearer_on_notifications(

        self,

        api_client: TestClient,

tests/unit/test_qod_service.py

+3 −0
Changes for tests/unit/test_qod_service.py: 3 added lines, 0 removed lines.
Original line number Diff line number Diff line
@@ -327,6 +327,9 @@ class TestCreateSession: 
        )

        assert decrypt_secret(credential.encrypted_token, CALLBACK_KEY) == RAW_TOKEN

        assert RAW_TOKEN.encode() not in credential.encrypted_token

        assert info.webhook is not None

        assert info.webhook.notificationUrl == "https://client.example.com/cb"

        assert info.webhook.notificationAuthToken is None

        saved_rows = [

            *operation_repo.rows.values(),

            *qod_session_repo.rows.values(),