feat: implement validation for IPv4 and IPv6 addresses in schemas and enhance... (73874c2b) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/adapters/http/srm_client.py

+28 −17

Original line number	Diff line number	Diff line
		@@ -56,6 +56,18 @@ _QUERY_PROBLEMS: dict[str, tuple[ErrorCode, str]] = {

		_UNKNOWN_QUERY_PROBLEM = _QUERY_PROBLEMS[f"{_PROBLEM_NS}service-not-applicable"]

		_PROBLEM_MEDIA_TYPE = "application/problem+json"


		def _problem_document(response: httpx.Response) -> dict[str, Any] \| None:
		if _PROBLEM_MEDIA_TYPE not in response.headers.get("content-type", ""):
		return None
		try:
		body = response.json()
		except ValueError: # JSONDecodeError -- truncated or mislabelled body
		return None
		return body if isinstance(body, dict) else None


		class SRMClient:
		def __init__(self) -> None:
		@@ -89,16 +101,15 @@ class SRMClient:
		headers=headers,
		)

		if response.status_code == 404:
		# Only the query calls translate SRM problems into CAMARA codes; every other call
		# keeps the plain 404 handling it has always had.
		problem = _problem_document(response) if problems is not None else None

		if response.status_code == 404 and (problems is None or problem is not None):
		log.warning("SRM resource not found", url=url)
		raise NotFoundException(message="Resource not found")

		if response.status_code == 422 and problems is not None:
		try:
		problem = response.json()
		except ValueError: # JSONDecodeError -- HTML error page, empty body, truncation
		problem = None
		if isinstance(problem, dict):
		if response.status_code == 422 and problems is not None and problem is not None:
		problem_type = problem.get("type")
		key = problem_type if isinstance(problem_type, str) else ""
		error_code, message = problems.get(key, _UNKNOWN_QUERY_PROBLEM)

src/open_exposure_gateway/api/camara/common.py

+14 −0

Original line number	Diff line number	Diff line
		import re
		from ipaddress import IPv4Address, IPv6Address
		from typing import Annotated, Optional

		from fastapi import Header
		@@ -15,3 +16,16 @@ IdempotencyKeyHeader = Annotated[
		Optional[str],
		Header(alias="Idempotency-Key", max_length=128),
		]


		def validated_ipv4(value: Optional[str]) -> Optional[str]:
		if value is not None:
		IPv4Address(value)
		return value


		def validated_ipv6(value: Optional[str]) -> Optional[str]:
		"""CAMARA's `DeviceIpv6Address` is `format: ipv6`; same reasoning as `validated_ipv4`."""
		if value is not None:
		IPv6Address(value)
		return value

src/open_exposure_gateway/api/camara/location_retrieval/v0_5_0/schemas.py

+16 −2

Original line number	Diff line number	Diff line
		from enum import StrEnum
		from typing import Annotated, Literal, Optional, Union

		from pydantic import AwareDatetime, BaseModel, Field, model_validator
		from pydantic import AwareDatetime, BaseModel, Field, field_validator, model_validator

		from open_exposure_gateway.api.camara.common import validated_ipv4, validated_ipv6


		class AreaType(StrEnum):
		@@ -14,6 +16,11 @@ class DeviceIpv4Addr(BaseModel):
		privateAddress: Optional[str] = None
		publicPort: Optional[int] = Field(default=None, ge=0, le=65535)

		@field_validator("publicAddress", "privateAddress")
		@classmethod
		def _valid_ipv4(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv4(value)

		@model_validator(mode="after")
		def _require_public_plus_one(self) -> "DeviceIpv4Addr":
		if self.publicAddress is None:
		@@ -29,6 +36,11 @@ class Device(BaseModel):
		ipv4Address: Optional[DeviceIpv4Addr] = None
		ipv6Address: Optional[str] = None

		@field_validator("ipv6Address")
		@classmethod
		def _valid_ipv6(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv6(value)

		@model_validator(mode="after")
		def _require_at_least_one_identifier(self) -> "Device":
		if not any(
		@@ -58,7 +70,9 @@ class DeviceResponse(Device):

		class RetrievalLocationRequest(BaseModel):
		device: Optional[Device] = None
		maxAge: Optional[int] = None
		# CAMARA leaves maxAge unbounded; SRM's canonical max_age_seconds is >= 0 (§E.2), so a
		# negative value is the client's 400 rather than a downstream rejection.
		maxAge: Optional[int] = Field(default=None, ge=0)
		maxSurface: Optional[int] = Field(default=None, ge=1)

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

+13 −1

Original line number	Diff line number	Diff line
		@@ -2,7 +2,9 @@ from enum import StrEnum
		from typing import Literal, Optional
		from uuid import UUID

		from pydantic import BaseModel, Field, model_validator
		from pydantic import BaseModel, Field, field_validator, model_validator

		from open_exposure_gateway.api.camara.common import validated_ipv4, validated_ipv6


		class QosStatus(StrEnum):
		@@ -16,6 +18,11 @@ class DeviceIpv4Addr(BaseModel):
		privateAddress: Optional[str] = None
		publicPort: Optional[int] = Field(default=None, ge=0, le=65535)

		@field_validator("publicAddress", "privateAddress")
		@classmethod
		def _valid_ipv4(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv4(value)

		@model_validator(mode="after")
		def _require_public_plus_one(self) -> "DeviceIpv4Addr":
		if self.publicAddress is None:
		@@ -31,6 +38,11 @@ class Device(BaseModel):
		ipv4Address: Optional[DeviceIpv4Addr] = None
		ipv6Address: Optional[str] = None

		@field_validator("ipv6Address")
		@classmethod
		def _valid_ipv6(cls, value: Optional[str]) -> Optional[str]:
		return validated_ipv6(value)

		@model_validator(mode="after")
		def _require_at_least_one_identifier(self) -> "Device":
		if not any(

tests/unit/test_location_retrieval_flows.py

+11 −6

Original line number	Diff line number	Diff line
		@@ -121,16 +121,21 @@ class TestQueryTranslation:

		assert located_device.location_queries[0].parameters.max_age_seconds == 0

		def test_negative_max_age_is_accepted_and_forwarded(
		def test_negative_max_age_is_rejected_here(
		self, api_client: TestClient, located_device: FakeSRMClient
		) -> None:
		"""The vendored schema puts no minimum on maxAge -- deliberately, since maxSurface
		beside it does carry one. Rejecting a negative here would 400 a spec-conformant
		request; whether the network can honour it is SRM's judgement, not OEG's."""
		"""The vendored schema puts no minimum on maxAge, but SRM's canonical
		`max_age_seconds` is `>= 0` (Interface Contract §E.2) and SRM enforces it.

		Forwarding a negative would buy CAMARA schema-conformance and pay for it with a 503:
		SRM rejects the body, and a rejected body is a fault OEG cannot report as anything
		useful. A 400 naming the field is the answer the caller can act on, so OEG is
		deliberately stricter than the vendored schema on this one field."""
		response = api_client.post(RETRIEVE, json={"device": {"phoneNumber": PHONE}, "maxAge": -5})

		assert response.status_code == 200
		assert located_device.location_queries[0].parameters.max_age_seconds == -5
		assert response.status_code == 400
		assert response.json()["code"] == "INVALID_ARGUMENT"
		assert located_device.location_queries == []

		def test_absent_max_age_stays_absent(
		self, api_client: TestClient, located_device: FakeSRMClient