fix(TI): reject device-scoped requests without an IPv4 address (cc281ba9) · Commits · OOP / code / open-exposure-gateway

src/open_exposure_gateway/application/services/traffic_influence_service.py

+1 −2

Changes for src/open_exposure_gateway/application/services/traffic_influence_service.py: 1 added line, 2 removed lines.
Original line number	Diff line number	Diff line
		@@ -261,8 +261,7 @@ class TrafficInfluenceService:
		error_code=ErrorCode.MISSING_IDENTIFIER,
		message="The device cannot be identified.",
		)
		if device.networkAccessIdentifier is not None:
		if not (device.phoneNumber or device.ipv4Address or device.ipv6Address):
		if device.ipv4Address is None:
		raise UnprocessableEntityException(
		error_code=ErrorCode.SERVICE_NOT_APPLICABLE,
		message="The service is not available for the provided identifier.",

tests/unit/test_traffic_influence_camara_schema.py

+9 −1

Changes for tests/unit/test_traffic_influence_camara_schema.py: 9 added lines, 1 removed line.
Original line number	Diff line number	Diff line
		@@ -74,7 +74,10 @@ class TestTrafficInfluenceCamaraSchema:
		def test_device_create_response_matches_schema(self, api_client: TestClient) -> None:
		response = api_client.post(
		f"{TI_BASE}/traffic-influence-devices",
		json={**TI_BODY, "device": {"phoneNumber": "+123456789"}},
		json={
		**TI_BODY,
		"device": {"ipv4Address": {"publicAddress": "84.125.93.10", "publicPort": 59765}},
		},
		)
		assert response.status_code == 201
		_assert_matches(response.json(), "TrafficInfluence")
		@@ -88,6 +91,11 @@ class TestTrafficInfluenceCamaraSchema:
		"SERVICE_NOT_APPLICABLE",
		id="network-access-identifier-only",
		),
		pytest.param(
		{**TI_BODY, "device": {"phoneNumber": "+123456789"}},
		"SERVICE_NOT_APPLICABLE",
		id="no-ipv4",
		),
		],
		)
		def test_device_identifier_errors_match_generic422(

tests/unit/test_traffic_influence_flows.py

+35 −13

Changes for tests/unit/test_traffic_influence_flows.py: 35 added lines, 13 removed lines.
Original line number	Diff line number	Diff line
		@@ -1726,7 +1726,8 @@ class TestTrafficInfluenceUnimplementedOperations:
		assert response.json()["code"] == "NOT_IMPLEMENTED"


		DEVICE_TI_BODY: dict[str, Any] = {**TI_BODY, "device": {"phoneNumber": "+123456789"}}
		DEVICE_IPV4: dict[str, Any] = {"publicAddress": "84.125.93.10", "publicPort": 59765}
		DEVICE_TI_BODY: dict[str, Any] = {**TI_BODY, "device": {"ipv4Address": DEVICE_IPV4}}


		class TestTrafficInfluenceDeviceCreateFlow:
		@@ -1798,13 +1799,16 @@ class TestTrafficInfluenceDeviceCreateFlow:
		) -> None:
		body = {
		**TI_BODY,
		"device": {"phoneNumber": "+123456789", "networkAccessIdentifier": "u@domain.com"},
		"device": {"ipv4Address": DEVICE_IPV4, "networkAccessIdentifier": "u@domain.com"},
		}
		response = api_client.post(f"{TI_BASE}/traffic-influence-devices", json=body)

		assert response.status_code == 201
		activate = [p for s, p in fake_bus.published if s == Subject.TASK_ACTIVATE][0]
		assert activate["network_capability"]["target"]["device"] == {"phone_number": "+123456789"}
		assert activate["network_capability"]["target"]["device"] == {
		"ipv4": "84.125.93.10",
		"ipv4_port": 59765,
		}

		def test_any_user_create_sends_no_device_target(
		self, api_client: TestClient, fake_bus: FakeDataBus
		@@ -1826,7 +1830,10 @@ class TestTrafficInfluenceDeviceCreateFlow:
		operations = list(operation_repo.rows.values())
		assert len(operations) == 1
		assert operations[0].operation_type == OperationType.TRAFFIC_INFLUENCE
		assert operations[0].metadata["request"]["device"]["phoneNumber"] == "+123456789"
		assert operations[0].metadata["request"]["device"]["ipv4Address"] == {
		**DEVICE_IPV4,
		"privateAddress": None,
		}
		stored = traffic_influence_repo.rows[traffic_influence_id]
		assert stored.state == TrafficInfluenceState.ORDERED
		assert stored.operation_id == operations[0].operation_id
		@@ -1975,16 +1982,18 @@ class TestTrafficInfluenceDeviceValidation:
		@pytest.mark.parametrize(
		"device",
		[
		pytest.param({"phoneNumber": "+123456789"}, id="phone"),
		pytest.param(
		{"ipv4Address": {"publicAddress": "84.125.93.10", "publicPort": 59765}},
		id="ipv4-public-and-port",
		),
		pytest.param({"ipv4Address": DEVICE_IPV4}, id="ipv4-public-and-port"),
		pytest.param(
		{"ipv4Address": {"publicAddress": "84.125.93.10", "privateAddress": "10.0.0.5"}},
		id="ipv4-public-and-private",
		),
		pytest.param({"ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"}, id="ipv6"),
		pytest.param(
		{"phoneNumber": "+123456789", "ipv4Address": DEVICE_IPV4}, id="phone-and-ipv4"
		),
		pytest.param(
		{"ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344", "ipv4Address": DEVICE_IPV4},
		id="ipv6-and-ipv4",
		),
		],
		)
		def test_each_supported_identifier_is_accepted(
		@@ -2011,15 +2020,28 @@ class TestTrafficInfluenceDeviceValidation:
		assert not fake_bus.published
		assert operation_repo.rows == {}

		def test_network_access_identifier_alone_is_not_applicable(
		@pytest.mark.parametrize(
		"device",
		[
		pytest.param({"phoneNumber": "+123456789"}, id="phone-only"),
		pytest.param({"ipv6Address": "2001:db8:85a3:8d3:1319:8a2e:370:7344"}, id="ipv6-only"),
		pytest.param(
		{"phoneNumber": "+123456789", "ipv6Address": "2001:db8::1"}, id="phone-and-ipv6"
		),
		pytest.param({"networkAccessIdentifier": "123456789@domain.com"}, id="nai-only"),
		],
		)
		def test_device_without_ipv4_is_not_applicable(
		self,
		api_client: TestClient,
		fake_bus: FakeDataBus,
		operation_repo: FakeOperationRepository,
		device: dict[str, Any],
		) -> None:
		"""SRM can only realize a device-scoped policy from the device's IPv4 address, so
		any other identifier is refused up front instead of a 201 that always errors."""
		response = api_client.post(
		f"{TI_BASE}/traffic-influence-devices",
		json={**TI_BODY, "device": {"networkAccessIdentifier": "123456789@domain.com"}},
		f"{TI_BASE}/traffic-influence-devices", json={**TI_BODY, "device": device}
		)

		assert response.status_code == 422