fix: return INVALID_TOKEN for non-bearer sinkCredential and reject expired tokens (eebb0338) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/schemas.py

+15 −2

Changes for src/open_exposure_gateway/api/camara/quality_on_demand/v1_1_0/schemas.py: 15 added lines, 2 removed lines.
Original line number	Diff line number	Diff line
		from datetime import datetime, timezone
		from enum import StrEnum
		from typing import Literal, Optional
		from typing import Optional
		from uuid import UUID

		from pydantic import AwareDatetime, BaseModel, Field, field_validator, model_validator
		@@ -120,7 +121,19 @@ class SinkCredential(BaseModel):
		credentialType: CredentialType
		accessToken: Optional[str] = Field(default=None, repr=False)
		accessTokenExpiresUtc: Optional[AwareDatetime] = None
		accessTokenType: Optional[Literal["bearer"]] = None
		accessTokenType: Optional[str] = None

		@field_validator("accessTokenExpiresUtc")
		@classmethod
		def _expires_in_future(cls, value: datetime \| None) -> datetime \| None:
		if value is not None and value <= datetime.now(timezone.utc):
		raise ValueError("accessTokenExpiresUtc must be in the future")
		return value

		@field_validator("accessTokenType")
		@classmethod
		def _lowercase_token_type(cls, value: str \| None) -> str \| None:
		return value.lower() if value is not None else None

		@model_validator(mode="after")
		def _require_access_token_fields(self) -> "SinkCredential":

src/open_exposure_gateway/application/services/quality_on_demand_service.py

+5 −0

Changes for src/open_exposure_gateway/application/services/quality_on_demand_service.py: 5 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -169,6 +169,11 @@ class QualityOnDemandService:
		message="Only Access token is supported",
		error_code=ErrorCode.INVALID_CREDENTIAL,
		)
		if sink_credential.accessTokenType != "bearer":
		raise BadRequestException(
		message="Only bearer token is supported",
		error_code=ErrorCode.INVALID_TOKEN,
		)
		if self._callback_secret_key is None:
		raise NotImplementedException(
		"sinkCredential is not supported by this gateway: no callback secret "

tests/unit/test_qod_flows.py

+31 −0

Changes for tests/unit/test_qod_flows.py: 31 added lines, 0 removed lines.
Original line number	Diff line number	Diff line
		@@ -418,6 +418,15 @@ class TestSinkCredentialWithKeyFlow:
		def callback_secret_key(self) -> bytes:
		return bytes(range(32))

		@pytest.mark.parametrize("token_type", ["Bearer", "BEARER"])
		def test_token_type_is_case_insensitive(self, api_client: TestClient, token_type: str) -> None:
		body = {
		**AUTH_SINK_BODY,
		"sinkCredential": {**AUTH_SINK_BODY["sinkCredential"], "accessTokenType": token_type},
		}
		response = api_client.post(f"{QOD_BASE}/sessions", json=body)
		assert response.status_code == 201

		def test_create_response_omits_the_token(self, api_client: TestClient) -> None:
		response = api_client.post(f"{QOD_BASE}/sessions", json=AUTH_SINK_BODY)
		assert response.status_code == 201
		@@ -529,6 +538,28 @@ class TestQodSessionValidation:
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_CREDENTIAL"

		def test_non_bearer_token_type_is_invalid_token(self, api_client: TestClient) -> None:
		body = {
		**AUTH_SINK_BODY,
		"sinkCredential": {**AUTH_SINK_BODY["sinkCredential"], "accessTokenType": "mac"},
		}
		response = api_client.post(f"{QOD_BASE}/sessions", json=body)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_TOKEN"

		def test_past_token_expiry_is_rejected(self, api_client: TestClient) -> None:
		body = {
		**AUTH_SINK_BODY,
		"sinkCredential": {
		**AUTH_SINK_BODY["sinkCredential"],
		"accessTokenExpiresUtc": "2020-01-01T00:00:00Z",
		},
		}
		response = api_client.post(f"{QOD_BASE}/sessions", json=body)
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"
		assert SINK_ACCESS_TOKEN not in response.text

		def test_phone_number_without_plus_is_rejected(self, api_client: TestClient) -> None:
		body = {**SESSION_BODY, "device": {"phoneNumber": "123456789"}}
		response = api_client.post(f"{QOD_BASE}/sessions", json=body)