fix(perforce): Extend the hostname checks

src/sentry/integrations/perforce/client.py

@@ -29,6 +29,7 @@
 from sentry.integrations.source_code_management.repository import RepositoryClient
 from sentry.models.pullrequest import PullRequest, PullRequestComment
 from sentry.models.repository import Repository
+from sentry.net.socket import is_safe_hostname
 from sentry.shared_integrations.exceptions import ApiError, ApiUnauthorized, IntegrationError
 
 logger = logging.getLogger(__name__)
@@ -68,6 +69,10 @@ def validate_p4port_transport(p4port: str) -> None:
         raise InvalidP4Port("Invalid P4PORT transport. Only tcp and ssl transports are allowed.")
     if not host:
         raise InvalidP4Port("P4PORT must include a host, e.g. ssl:perforce.example.com:1666.")
+    if not is_safe_hostname(host):
+        raise InvalidP4Port(
+            f"P4PORT host could not be resolved or is not an allowed address: {host}"
+        )
 
 
 class P4ChangeInfo(TypedDict):

src/sentry/integrations/perforce/integration.py

@@ -605,6 +605,16 @@ def update_organization_config(self, data: Mapping[str, Any]) -> None:
         """
         from sentry.integrations.services.integration import integration_service
 
+        data = dict(data)
+
+        if "p4port" in data:
+            p4port = str(data["p4port"]).strip().rstrip("/")
+            try:
+                validate_p4port_transport(p4port)
+            except InvalidP4Port as e:
+                raise IntegrationError(str(e))
+            data["p4port"] = p4port
+
         # Update integration metadata with new values
         metadata = dict(self.model.metadata)  # Create a mutable copy
         metadata.update(data)  # Only update fields present in data

tests/sentry/integrations/perforce/test_client.py

@@ -40,6 +40,14 @@ def setUp(self) -> None:
             integration=self.integration, org_integration=self.org_integration
         )
 
+        # These tests mock the P4 connection with placeholder hosts that don't
+        # resolve; bypass the SSRF host check (covered in test_integration.py).
+        patcher = mock.patch(
+            "sentry.integrations.perforce.client.is_safe_hostname", return_value=True
+        )
+        patcher.start()
+        self.addCleanup(patcher.stop)
+
     @mock.patch("sentry.integrations.perforce.client.P4")
     def test_get_blame_for_files_success(self, mock_p4_class):
         """Test get_blame_for_files returns commit info for files"""

tests/sentry/integrations/perforce/test_integration.py

@@ -14,7 +14,7 @@
 from sentry.integrations.pipeline import IntegrationPipeline
 from sentry.models.repository import Repository
 from sentry.organizations.services.organization.serial import serialize_rpc_organization
-from sentry.shared_integrations.exceptions import ApiError, ApiUnauthorized
+from sentry.shared_integrations.exceptions import ApiError, ApiUnauthorized, IntegrationError
 from sentry.silo.base import SiloMode
 from sentry.testutils.cases import APITestCase, IntegrationTestCase
 from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
@@ -258,7 +258,8 @@ def test_validate_web_url_adds_scheme(self) -> None:
         )
         assert serializer.validate_web_url("http://swarm.example.com") == "http://swarm.example.com"
 
-    def test_serializer_normalizes_schemeless_web_url_end_to_end(self) -> None:
+    @patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+    def test_serializer_normalizes_schemeless_web_url_end_to_end(self, mock_safe_hostname) -> None:
         """
         Exercise the full serializer, not just validate_web_url directly. With a URLField
         the field-level URLValidator rejected schemeless input before validate_web_url ran,
@@ -510,7 +511,8 @@ def test_validate_p4port_transport_rejects_invalid(self) -> None:
             with pytest.raises(InvalidP4Port):
                 validate_p4port_transport(value)
 
-    def test_validate_p4port_transport_allows_host_port(self) -> None:
+    @patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+    def test_validate_p4port_transport_allows_host_port(self, mock_safe_hostname) -> None:
         for value in (
             "perforce.example.com:1666",
             "test-host.example.com:1666",
@@ -520,7 +522,8 @@ def test_validate_p4port_transport_allows_host_port(self) -> None:
         ):
             validate_p4port_transport(value)
 
-    def test_validate_p4port_transport_allows_every_transport(self) -> None:
+    @patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+    def test_validate_p4port_transport_allows_every_transport(self, mock_safe_hostname) -> None:
         for transport in (
             "tcp",
             "tcp4",
@@ -548,11 +551,19 @@ def test_validate_p4port_transport_rejects_disallowed_transport(self) -> None:
             with pytest.raises(InvalidP4Port):
                 validate_p4port_transport(f"{transport}:perforce.example.com:1666")
 
-    def test_serializer_accepts_valid_p4port(self) -> None:
+    @patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+    def test_serializer_accepts_valid_p4port(self, mock_safe_hostname) -> None:
         serializer = PerforceInstallationSerializer(data=self._base_payload())
         assert serializer.is_valid(), serializer.errors
         assert serializer.validated_data["p4port"] == "ssl:perforce.example.com:1666"
 
+    def test_serializer_rejects_internal_host(self) -> None:
+        payload = self._base_payload()
+        payload["p4port"] = "tcp:169.254.169.254:1666"
+        serializer = PerforceInstallationSerializer(data=payload)
+        assert not serializer.is_valid()
+        assert "p4port" in serializer.errors
+
     def test_client_connect_rejects_invalid_p4port_metadata(self) -> None:
         with assume_test_silo_mode(SiloMode.CELL):
             integration = self.create_integration(
@@ -568,6 +579,52 @@ def test_client_connect_rejects_invalid_p4port_metadata(self) -> None:
             with client._connect():
                 pass
 
+    def test_client_connect_rejects_internal_host(self) -> None:
+        with assume_test_silo_mode(SiloMode.CELL):
+            integration = self.create_integration(
+                organization=self.organization,
+                provider="perforce",
+                name="Perforce",
+                external_id="perforce-local",
+                metadata={"p4port": "tcp:169.254.169.254:1666", "user": "u", "password": "p"},
+            )
+        installation = integration.get_installation(self.organization.id)
+        client = installation.get_client()
+        with pytest.raises(ApiError):
+            with client._connect():
+                pass
+
+    def _installation_for_update(self, external_id: str) -> PerforceIntegration:
+        with assume_test_silo_mode(SiloMode.CELL):
+            integration = self.create_integration(
+                organization=self.organization,
+                provider="perforce",
+                name="Perforce",
+                external_id=external_id,
+                metadata={"p4port": "ssl:perforce.example.com:1666", "user": "u", "password": "p"},
+            )
+        installation = integration.get_installation(self.organization.id)
+        assert isinstance(installation, PerforceIntegration)
+        return installation
+
+    def test_update_organization_config_rejects_disallowed_transport(self) -> None:
+        installation = self._installation_for_update("perforce-update-invalid")
+        with pytest.raises(IntegrationError):
+            installation.update_organization_config({"p4port": "demo:test"})
+
+    def test_update_organization_config_rejects_internal_host(self) -> None:
+        installation = self._installation_for_update("perforce-update-local")
+        with pytest.raises(IntegrationError):
+            installation.update_organization_config({"p4port": "tcp:169.254.169.254:1666"})
+
+    @patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+    def test_update_organization_config_accepts_and_normalizes_p4port(
+        self, mock_safe_hostname
+    ) -> None:
+        installation = self._installation_for_update("perforce-update-ok")
+        installation.update_organization_config({"p4port": "tcp:perforce.example.com:1666/"})
+        assert installation.get_config_data()["p4port"] == "tcp:perforce.example.com:1666"
+
 
 class PerforceIntegrationCodeMappingTest(IntegrationTestCase):
     """Tests for Perforce integration with code mappings"""
@@ -931,6 +988,9 @@ class PerforceApiPipelineTest(APITestCase):
     def setUp(self):
         super().setUp()
         self.login_as(self.user)
+        patcher = patch("sentry.integrations.perforce.client.is_safe_hostname", return_value=True)
+        patcher.start()
+        self.addCleanup(patcher.stop)
 
     def _get_pipeline_url(self) -> str:
         return reverse(