Merge branch 'main' into task-fix-suspension

Author: jzbahrai

app/models.py

@@ -1081,6 +1081,7 @@ class ApiKeyPermission(StrEnum):
     """
 
     MANAGE_TEMPLATES = "manage_templates"
+    MANAGE_REPORTS = "manage_reports"
 
 
 API_KEY_PERMISSION_TYPES: frozenset[str] = frozenset(p.value for p in ApiKeyPermission)

app/service/callback_rest.py

@@ -1,3 +1,6 @@
+import ipaddress
+from urllib.parse import urlparse
+
 from flask import Blueprint, jsonify, request
 from sqlalchemy.exc import SQLAlchemyError
 
@@ -36,6 +39,7 @@
 def create_service_inbound_api(service_id):
     data = request.get_json()
     validate(data, create_service_callback_api_schema)
+    _validate_not_localhost(data.get("url"))
     data["service_id"] = service_id
     inbound_api = ServiceInboundApi(**data)
     try:
@@ -50,6 +54,7 @@ def create_service_inbound_api(service_id):
 def update_service_inbound_api(service_id, inbound_api_id):
     data = request.get_json()
     validate(data, update_service_callback_api_schema)
+    _validate_not_localhost(data.get("url"))
 
     to_update = get_service_inbound_api(inbound_api_id, service_id)
 
@@ -85,6 +90,7 @@ def remove_service_inbound_api(service_id, inbound_api_id):
 def create_service_callback_api(service_id):
     data = request.get_json()
     validate(data, create_service_callback_api_schema)
+    _validate_not_localhost(data.get("url"))
     data["service_id"] = service_id
     data["callback_type"] = DELIVERY_STATUS_CALLBACK_TYPE
     callback_api = ServiceCallbackApi(**data)
@@ -100,6 +106,7 @@ def create_service_callback_api(service_id):
 def update_service_callback_api(service_id, callback_api_id):
     data = request.get_json()
     validate(data, update_service_callback_api_schema)
+    _validate_not_localhost(data.get("url"))
 
     to_update = get_service_callback_api(callback_api_id, service_id)
 
@@ -172,3 +179,25 @@ def handle_sql_error(e, table_name):
         return jsonify(result="error", message="No result found"), 404
     else:
         raise e
+
+
+def _validate_not_localhost(url):
+    """Reject callback URLs that reference localhost or loopback addresses."""
+    if url:
+        try:
+            parsed = urlparse(url)
+            hostname = (parsed.hostname or "").rstrip(".")
+            is_loopback = hostname == "localhost" or hostname.endswith(".localhost")
+            if not is_loopback:
+                try:
+                    is_loopback = ipaddress.ip_address(hostname).is_loopback
+                except ValueError:
+                    # Hostname is not an IP literal (for example, a normal DNS name);
+                    # keep `is_loopback` as determined by localhost-name checks above.
+                    pass
+            if is_loopback:
+                raise InvalidRequest("Callback URL must not point to localhost", status_code=400)
+        except InvalidRequest:
+            raise
+        except Exception as e:
+            raise InvalidRequest("Invalid callback URL", status_code=400) from e

tests/app/dao/test_api_key_dao.py

@@ -16,7 +16,7 @@
     update_compromised_api_key_info,
     update_last_used_api_key,
 )
-from app.models import KEY_TYPE_NORMAL, ApiKey
+from app.models import API_KEY_PERMISSION_TYPES, KEY_TYPE_NORMAL, ApiKey, ApiKeyPermission
 from tests.app.db import create_api_key
 from tests.conftest import set_signer_secret_key
 
@@ -286,6 +286,13 @@ def test_can_persist_and_round_trip_known_permission(self, sample_service):
         fetched = ApiKey.query.get(api_key.id)
         assert fetched.permissions == ["manage_templates"]
 
+    def test_can_persist_and_round_trip_manage_reports_permission(self, sample_service):
+        assert "manage_reports" in API_KEY_PERMISSION_TYPES
+        api_key = self._make_key(sample_service, permissions=[ApiKeyPermission.MANAGE_REPORTS])
+        fetched = ApiKey.query.get(api_key.id)
+        assert fetched.permissions == ["manage_reports"]
+        assert fetched.has_permission("manage_reports") is True
+
     def test_unknown_permission_is_rejected(self, sample_service):
         with pytest.raises(ValueError, match="Invalid api key permission"):
             self._make_key(sample_service, permissions=["not_a_real_permission"])

tests/app/service/test_callback_rest.py

@@ -3,6 +3,7 @@
 import pytest
 
 from app.models import ServiceCallbackApi, ServiceInboundApi
+from app.service.callback_rest import _validate_not_localhost
 from tests.app.db import create_service_callback_api, create_service_inbound_api
 
 
@@ -217,3 +218,85 @@ def test_suspend_callback_api(self, admin_request, sample_service, suspend_unsus
         callback = ServiceCallbackApi.query.get(service_callback_api.id)
         assert callback.is_suspended is suspend_unsuspend
         assert callback.updated_by_id == sample_service.users[0].id
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://localhost/callback",
+        "https://localhost./callback",
+        "https://localhost:8080/callback",
+        "https://127.0.0.1/callback",
+        "https://127.0.0.2/callback",
+        "https://127.255.255.255/callback",
+        "https://127.0.0.1:443/callback",
+        "https://[::1]/callback",
+        "https://[0:0:0:0:0:0:0:1]/callback",
+        "https://sub.localhost/callback",
+    ],
+)
+def test_validate_not_localhost_rejects_localhost_urls(url):
+    from app.errors import InvalidRequest
+
+    with pytest.raises(InvalidRequest):
+        _validate_not_localhost(url)
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://example.com/callback",
+        "https://my-service.gc.ca/callback",
+        None,
+    ],
+)
+def test_validate_not_localhost_allows_valid_urls(url):
+    _validate_not_localhost(url)  # should not raise
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://localhost/callback",
+        "https://127.0.0.1/callback",
+        "https://127.0.0.2/callback",
+        "https://[::1]/callback",
+    ],
+)
+def test_create_inbound_api_rejects_localhost(admin_request, sample_service, url):
+    data = {
+        "url": url,
+        "bearer_token": "some-unique-string",
+        "updated_by_id": str(sample_service.users[0].id),
+    }
+    resp_json = admin_request.post(
+        "service_callback.create_service_inbound_api",
+        service_id=sample_service.id,
+        _data=data,
+        _expected_status=400,
+    )
+    assert "localhost" in resp_json["message"].lower()
+
+
+@pytest.mark.parametrize(
+    "url",
+    [
+        "https://localhost/callback",
+        "https://127.0.0.1/callback",
+        "https://127.0.0.2/callback",
+        "https://[::1]/callback",
+    ],
+)
+def test_create_callback_api_rejects_localhost(admin_request, sample_service, url):
+    data = {
+        "url": url,
+        "bearer_token": "some-unique-string",
+        "updated_by_id": str(sample_service.users[0].id),
+    }
+    resp_json = admin_request.post(
+        "service_callback.create_service_callback_api",
+        service_id=sample_service.id,
+        _data=data,
+        _expected_status=400,
+    )
+    assert "localhost" in resp_json["message"].lower()