API - Notification URLs werent always being validated (#3812)

changedetectionio/api/Tags.py

@@ -96,6 +96,16 @@ class Tag(Resource):
         if not tag:
             abort(404, message='No tag exists with the UUID of {}'.format(uuid))
 
+        # Validate notification_urls if provided
+        if 'notification_urls' in request.json:
+            from wtforms import ValidationError
+            from changedetectionio.api.Notifications import validate_notification_urls
+            try:
+                notification_urls = request.json.get('notification_urls', [])
+                validate_notification_urls(notification_urls)
+            except ValidationError as e:
+                return str(e), 400
+
         tag.update(request.json)
         self.datastore.needs_write_urgent = True
 

changedetectionio/api/Watch.py

@@ -140,6 +140,16 @@ class Watch(Resource):
         if validation_error:
             return validation_error, 400
 
+        # Validate notification_urls if provided
+        if 'notification_urls' in request.json:
+            from wtforms import ValidationError
+            from changedetectionio.api.Notifications import validate_notification_urls
+            try:
+                notification_urls = request.json.get('notification_urls', [])
+                validate_notification_urls(notification_urls)
+            except ValidationError as e:
+                return str(e), 400
+
         # XSS etc protection - validate URL if it's being updated
         if 'url' in request.json:
             new_url = request.json.get('url')
@@ -444,6 +454,16 @@ class CreateWatch(Resource):
         if validation_error:
             return validation_error, 400
 
+        # Validate notification_urls if provided
+        if 'notification_urls' in json_data:
+            from wtforms import ValidationError
+            from changedetectionio.api.Notifications import validate_notification_urls
+            try:
+                notification_urls = json_data.get('notification_urls', [])
+                validate_notification_urls(notification_urls)
+            except ValidationError as e:
+                return str(e), 400
+
         extras = copy.deepcopy(json_data)
 
         # Because we renamed 'tag' to 'tags' but don't want to change the API (can do this in v2 of the API)

changedetectionio/tests/test_api_notification_urls_validation.py

@@ -0,0 +1,166 @@
+#!/usr/bin/env python3
+
+"""
+Test notification_urls validation in Watch and Tag API endpoints.
+Ensures that invalid AppRise URLs are rejected when setting notification_urls.
+
+Valid AppRise notification URLs use specific protocols like:
+- posts://example.com - POST to HTTP endpoint
+- gets://example.com - GET to HTTP endpoint
+- mailto://user@example.com - Email
+- slack://token/channel - Slack
+- discord://webhook_id/webhook_token - Discord
+- etc.
+
+Invalid notification URLs:
+- https://example.com - Plain HTTPS is NOT a valid AppRise notification protocol
+- ftp://example.com - FTP is NOT a valid AppRise notification protocol
+- Plain URLs without proper AppRise protocol prefix
+"""
+
+from flask import url_for
+import json
+
+
+def test_watch_notification_urls_validation(client, live_server, measure_memory_usage, datastore_path):
+    """Test that Watch PUT/POST endpoints validate notification_urls."""
+    api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
+
+    # Test 1: Create a watch with valid notification URLs
+    valid_urls = ["posts://example.com/notify1", "posts://example.com/notify2"]
+    res = client.post(
+        url_for("createwatch"),
+        data=json.dumps({
+            "url": "https://example.com",
+            "notification_urls": valid_urls
+        }),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 201, "Should accept valid notification URLs on watch creation"
+    watch_uuid = res.json['uuid']
+
+    # Verify the notification URLs were saved
+    res = client.get(
+        url_for("watch", uuid=watch_uuid),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert set(res.json['notification_urls']) == set(valid_urls), "Valid notification URLs should be saved"
+
+    # Test 2: Try to create a watch with invalid notification URLs (https:// is not valid)
+    invalid_urls = ["https://example.com/webhook"]
+    res = client.post(
+        url_for("createwatch"),
+        data=json.dumps({
+            "url": "https://example.com",
+            "notification_urls": invalid_urls
+        }),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject https:// notification URLs (not a valid AppRise protocol)"
+    assert b"is not a valid AppRise URL" in res.data, "Should provide AppRise validation error message"
+
+    # Test 2b: Also test other invalid protocols
+    invalid_urls_ftp = ["ftp://not-apprise-url"]
+    res = client.post(
+        url_for("createwatch"),
+        data=json.dumps({
+            "url": "https://example.com",
+            "notification_urls": invalid_urls_ftp
+        }),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject ftp:// notification URLs"
+    assert b"is not a valid AppRise URL" in res.data, "Should provide AppRise validation error message"
+
+    # Test 3: Update watch with valid notification URLs
+    new_valid_urls = ["posts://newserver.com"]
+    res = client.put(
+        url_for("watch", uuid=watch_uuid),
+        data=json.dumps({"notification_urls": new_valid_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 200, "Should accept valid notification URLs on watch update"
+
+    # Verify the notification URLs were updated
+    res = client.get(
+        url_for("watch", uuid=watch_uuid),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert res.json['notification_urls'] == new_valid_urls, "Valid notification URLs should be updated"
+
+    # Test 4: Try to update watch with invalid notification URLs (plain https:// not valid)
+    invalid_https_url = ["https://example.com/webhook"]
+    res = client.put(
+        url_for("watch", uuid=watch_uuid),
+        data=json.dumps({"notification_urls": invalid_https_url}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject https:// notification URLs on watch update"
+    assert b"is not a valid AppRise URL" in res.data, "Should provide AppRise validation error message"
+
+    # Test 5: Update watch with non-list notification_urls (caught by OpenAPI schema validation)
+    res = client.put(
+        url_for("watch", uuid=watch_uuid),
+        data=json.dumps({"notification_urls": "not-a-list"}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject non-list notification_urls"
+    assert b"OpenAPI validation failed" in res.data or b"Request body validation error" in res.data
+
+    # Test 6: Verify original URLs are preserved after failed update
+    res = client.get(
+        url_for("watch", uuid=watch_uuid),
+        headers={'x-api-key': api_key}
+    )
+    assert res.status_code == 200
+    assert res.json['notification_urls'] == new_valid_urls, "URLs should remain unchanged after validation failure"
+
+
+def test_tag_notification_urls_validation(client, live_server, measure_memory_usage, datastore_path):
+    """Test that Tag PUT endpoint validates notification_urls."""
+    from changedetectionio.model import Tag
+
+    api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token')
+    datastore = live_server.app.config['DATASTORE']
+
+    # Create a tag
+    tag_uuid = datastore.add_tag(title="Test Tag")
+    assert tag_uuid is not None
+
+    # Test 1: Update tag with valid notification URLs
+    valid_urls = ["posts://example.com/tag-notify"]
+    res = client.put(
+        url_for("tag", uuid=tag_uuid),
+        data=json.dumps({"notification_urls": valid_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 200, "Should accept valid notification URLs on tag update"
+
+    # Verify the notification URLs were saved
+    tag = datastore.data['settings']['application']['tags'][tag_uuid]
+    assert tag['notification_urls'] == valid_urls, "Valid notification URLs should be saved to tag"
+
+    # Test 2: Try to update tag with invalid notification URLs (https:// not valid)
+    invalid_urls = ["https://example.com/webhook"]
+    res = client.put(
+        url_for("tag", uuid=tag_uuid),
+        data=json.dumps({"notification_urls": invalid_urls}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject https:// notification URLs on tag update"
+    assert b"is not a valid AppRise URL" in res.data, "Should provide AppRise validation error message"
+
+    # Test 3: Update tag with non-list notification_urls (caught by OpenAPI schema validation)
+    res = client.put(
+        url_for("tag", uuid=tag_uuid),
+        data=json.dumps({"notification_urls": "not-a-list"}),
+        headers={'content-type': 'application/json', 'x-api-key': api_key}
+    )
+    assert res.status_code == 400, "Should reject non-list notification_urls"
+    assert b"OpenAPI validation failed" in res.data or b"Request body validation error" in res.data
+
+    # Test 4: Verify original URLs are preserved after failed update
+    tag = datastore.data['settings']['application']['tags'][tag_uuid]
+    assert tag['notification_urls'] == valid_urls, "URLs should remain unchanged after validation failure"