Which is across multiple lines
+ + + """ + with open(os.path.join(datastore_path, "endpoint-content.txt"), "w") as f: + f.write(test_return_data) + return None + + +def is_valid_uuid(val): + try: + uuid_module.UUID(str(val)) + return True + except ValueError: + return False + + +# ============================================================================ +# TIER 1: CRITICAL SECURITY TESTS +# ============================================================================ + +def test_api_path_traversal_in_uuids(client, live_server, measure_memory_usage, datastore_path): + """ + Test that path traversal attacks via UUID parameter are blocked. + Addresses CVE-like vulnerabilities where ../../../ in UUID could access arbitrary files. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Create a valid watch first + res = client.post( + url_for("createwatch"), + data=json.dumps({"url": test_url, "title": "Valid watch"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 201 + valid_uuid = res.json.get('uuid') + + # Test 1: Path traversal with ../../../ + res = client.get( + f"/api/v1/watch/../../etc/passwd", + headers={'x-api-key': api_key} + ) + assert res.status_code in [400, 404], "Path traversal should be rejected" + + # Test 2: Encoded path traversal + res = client.get( + "/api/v1/watch/..%2F..%2F..%2Fetc%2Fpasswd", + headers={'x-api-key': api_key} + ) + assert res.status_code in [400, 404], "Encoded path traversal should be rejected" + + # Test 3: Double-encoded path traversal + res = client.get( + "/api/v1/watch/%2e%2e%2f%2e%2e%2f%2e%2e%2f", + headers={'x-api-key': api_key} + ) + assert res.status_code in [400, 404], "Double-encoded traversal should be rejected" + + # Test 4: Try to access datastore file + res = client.get( + "/api/v1/watch/../url-watches.json", + headers={'x-api-key': api_key} + ) + assert res.status_code in [400, 404], "Access to datastore should be blocked" + + # Test 5: Null byte injection + res = client.get( + f"/api/v1/watch/{valid_uuid}%00.json", + headers={'x-api-key': api_key} + ) + # Should either work (ignoring null byte) or reject - but not crash + assert res.status_code in [200, 400, 404] + + # Test 6: DELETE with path traversal + res = client.delete( + "/api/v1/watch/../../datastore/url-watches.json", + headers={'x-api-key': api_key} + ) + assert res.status_code in [400, 404, 405], "DELETE with traversal should be blocked (405=method not allowed is also acceptable)" + + # Cleanup + client.delete(url_for("watch", uuid=valid_uuid), headers={'x-api-key': api_key}) + delete_all_watches(client) + + +def test_api_injection_via_headers_and_proxy(client, live_server, measure_memory_usage, datastore_path): + """ + Test that injection attacks via headers and proxy fields are properly sanitized. + Addresses XSS and injection vulnerabilities. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: XSS in headers + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "headers": { + "User-Agent": "", + "X-Custom": "'; DROP TABLE watches; --" + } + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Headers are metadata used for HTTP requests, not HTML rendering + # Storing them as-is is expected behavior + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + # Verify headers are stored (API returns JSON, not HTML, so no XSS risk) + res = client.get(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + assert res.status_code == 200 + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 2: Null bytes in headers + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "headers": {"X-Test": "value\x00null"} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should handle null bytes gracefully (reject or sanitize) + assert res.status_code in [201, 400] + + # Test 3: Malformed proxy string + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "proxy": "http://evil.com:8080@victim.com" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should reject invalid proxy format + assert res.status_code == 400 + + # Test 4: Control characters in notification title + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "notification_title": "Test\r\nInjected-Header: value" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept but sanitize control characters + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + delete_all_watches(client) + + +def test_api_large_payload_dos(client, live_server, measure_memory_usage, datastore_path): + """ + Test that excessively large payloads are rejected to prevent DoS. + Addresses memory leak issues found in changelog. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Huge ignore_text array + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "ignore_text": ["a" * 10000] * 100 # 1MB of data + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should either accept (with limits) or reject + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 2: Massive headers object + huge_headers = {f"X-Header-{i}": "x" * 1000 for i in range(100)} + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "headers": huge_headers + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should reject or truncate + assert res.status_code in [201, 400, 413] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 3: Huge browser_steps array + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "browser_steps": [ + {"operation": "click", "selector": "#test" * 1000, "optional_value": ""} + ] * 100 + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should reject or limit + assert res.status_code in [201, 400, 413] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 4: Extremely long title + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "title": "x" * 100000 # 100KB title + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should reject (exceeds maxLength: 5000) + assert res.status_code == 400 + + delete_all_watches(client) + + +def test_api_utf8_encoding_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test UTF-8 encoding edge cases that have caused bugs on Windows. + Addresses 18+ encoding bugs from changelog. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Unicode in title (should work) + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "title": "Test 中文 Ελληνικά 日本語 🔥" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 201 + watch_uuid = res.json.get('uuid') + + # Verify it round-trips correctly + res = client.get(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + assert res.status_code == 200 + assert "中文" in res.json.get('title') + + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 2: Unicode in URL query parameters + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url + "?search=日本語" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should handle URL encoding properly + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 3: Null byte in title (should be rejected or sanitized) + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "title": "Test\x00Title" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should handle gracefully + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 4: BOM (Byte Order Mark) in title + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "title": "\ufeffTest with BOM" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + delete_all_watches(client) + + +def test_api_concurrency_race_conditions(client, live_server, measure_memory_usage, datastore_path): + """ + Test concurrent API requests to detect race conditions. + Addresses 20+ concurrency bugs from changelog. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Create a watch + res = client.post( + url_for("createwatch"), + data=json.dumps({"url": test_url, "title": "Concurrency test"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 201 + watch_uuid = res.json.get('uuid') + wait_for_all_checks(client) + + # Test 1: Concurrent updates to same watch + # Note: Flask test client is not thread-safe, so we test sequential updates instead + # Real concurrency issues would be caught in integration tests with actual HTTP requests + results = [] + for i in range(10): + try: + r = client.put( + url_for("watch", uuid=watch_uuid), + data=json.dumps({"title": f"Title {i}"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + results.append(r.status_code) + except Exception as e: + results.append(str(e)) + + # All updates should succeed (200) without crashes + assert all(r == 200 for r in results), f"Some updates failed: {results}" + + # Test 2: Update while watch is being checked + # Queue a recheck + client.get( + url_for("watch", uuid=watch_uuid, recheck=True), + headers={'x-api-key': api_key} + ) + + # Immediately update it + res = client.put( + url_for("watch", uuid=watch_uuid), + data=json.dumps({"title": "Updated during check"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should succeed without error + assert res.status_code == 200 + + # Test 3: Delete watch that's being processed + # Create another watch + res = client.post( + url_for("createwatch"), + data=json.dumps({"url": test_url}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + watch_uuid2 = res.json.get('uuid') + + # Queue it for checking + client.get(url_for("watch", uuid=watch_uuid2, recheck=True), headers={'x-api-key': api_key}) + + # Immediately delete it + res = client.delete(url_for("watch", uuid=watch_uuid2), headers={'x-api-key': api_key}) + # Should succeed or return appropriate error + assert res.status_code in [204, 404, 400] + + # Cleanup + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + delete_all_watches(client) + + +# ============================================================================ +# TIER 2: IMPORTANT FUNCTIONALITY TESTS +# ============================================================================ + +def test_api_time_validation_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test time_between_check validation edge cases. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Zero interval + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "time_between_check_use_default": False, + "time_between_check": {"seconds": 0} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 400, "Zero interval should be rejected" + + # Test 2: Negative interval + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "time_between_check_use_default": False, + "time_between_check": {"seconds": -100} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 400, "Negative interval should be rejected" + + # Test 3: All fields null with use_default=false + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "time_between_check_use_default": False, + "time_between_check": {"weeks": None, "days": None, "hours": None, "minutes": None, "seconds": None} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 400, "All null intervals should be rejected when not using default" + + # Test 4: Extremely large interval (overflow risk) + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "time_between_check_use_default": False, + "time_between_check": {"weeks": 999999999} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should either accept (with limits) or reject + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 5: Valid minimal interval (should work) + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "time_between_check_use_default": False, + "time_between_check": {"seconds": 60} + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + assert res.status_code == 201 + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + delete_all_watches(client) + + +def test_api_browser_steps_validation(client, live_server, measure_memory_usage, datastore_path): + """ + Test browser_steps validation for invalid operations and structures. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Empty browser step + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "browser_steps": [ + {"operation": "", "selector": "", "optional_value": ""} + ] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept (empty is valid as null) + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 2: Invalid operation type + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "browser_steps": [ + {"operation": "invalid_operation", "selector": "#test", "optional_value": ""} + ] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept (validation happens at runtime) or reject + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 3: Missing required fields in browser step + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "browser_steps": [ + {"operation": "click"} # Missing selector and optional_value + ] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should be rejected due to schema validation + assert res.status_code == 400 + + # Test 4: Extra fields in browser step + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "browser_steps": [ + {"operation": "click", "selector": "#test", "optional_value": "", "extra_field": "value"} + ] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should be rejected due to additionalProperties: false + assert res.status_code == 400 + + delete_all_watches(client) + + +def test_api_queue_manipulation(client, live_server, measure_memory_usage, datastore_path): + """ + Test queue behavior under stress and edge cases. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Create many watches rapidly + watch_uuids = [] + for i in range(20): + res = client.post( + url_for("createwatch"), + data=json.dumps({"url": test_url, "title": f"Watch {i}"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + if res.status_code == 201: + watch_uuids.append(res.json.get('uuid')) + + assert len(watch_uuids) == 20, "Should be able to create 20 watches" + + # Test 2: Recheck all when watches exist + res = client.get( + url_for("createwatch", recheck_all='1'), + headers={'x-api-key': api_key}, + ) + # Should return success (200 or 202 for background processing) + assert res.status_code in [200, 202] + + # Test 3: Verify queue doesn't overflow with moderate load + # The app has MAX_QUEUE_SIZE = 5000, we're well below that + wait_for_all_checks(client) + + # Cleanup + for uuid in watch_uuids: + client.delete(url_for("watch", uuid=uuid), headers={'x-api-key': api_key}) + + delete_all_watches(client) + + +# ============================================================================ +# TIER 3: EDGE CASES & POLISH +# ============================================================================ + +def test_api_history_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test history API with invalid timestamps and edge cases. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Create watch and generate history + res = client.post( + url_for("createwatch"), + data=json.dumps({"url": test_url}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + watch_uuid = res.json.get('uuid') + wait_for_all_checks(client) + + # Test 1: Get history with invalid timestamp + res = client.get( + url_for("watchsinglehistory", uuid=watch_uuid, timestamp="invalid"), + headers={'x-api-key': api_key} + ) + assert res.status_code == 404, "Invalid timestamp should return 404" + + # Test 2: Future timestamp + res = client.get( + url_for("watchsinglehistory", uuid=watch_uuid, timestamp="9999999999"), + headers={'x-api-key': api_key} + ) + assert res.status_code == 404, "Future timestamp should return 404" + + # Test 3: Negative timestamp + res = client.get( + url_for("watchsinglehistory", uuid=watch_uuid, timestamp="-1"), + headers={'x-api-key': api_key} + ) + assert res.status_code == 404, "Negative timestamp should return 404" + + # Test 4: Diff with reversed timestamps (from > to) + # First get actual timestamps + res = client.get( + url_for("watchhistory", uuid=watch_uuid), + headers={'x-api-key': api_key} + ) + if len(res.json) >= 2: + timestamps = sorted(res.json.keys()) + # Try reversed order + res = client.get( + url_for("watchhistorydiff", uuid=watch_uuid, from_timestamp=timestamps[-1], to_timestamp=timestamps[0]), + headers={'x-api-key': api_key} + ) + # Should either work (show reverse diff) or return error + assert res.status_code in [200, 400] + + # Cleanup + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + delete_all_watches(client) + + +def test_api_notification_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test notification configuration edge cases. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Invalid notification URL + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "notification_urls": ["invalid://url", "ftp://test.com"] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept (apprise validates at runtime) or reject + assert res.status_code in [201, 400] + if res.status_code == 201: + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + # Test 2: Invalid notification format + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "notification_format": "invalid_format" + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should be rejected by schema + assert res.status_code == 400 + + # Test 3: Empty notification arrays + res = client.post( + url_for("createwatch"), + data=json.dumps({ + "url": test_url, + "notification_urls": [] + }), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept (empty is valid) + assert res.status_code == 201 + watch_uuid = res.json.get('uuid') + client.delete(url_for("watch", uuid=watch_uuid), headers={'x-api-key': api_key}) + + delete_all_watches(client) + + +def test_api_tag_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test tag/group API edge cases including XSS and path traversal. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + + # Test 1: Empty tag title + res = client.post( + url_for("tag"), + data=json.dumps({"title": ""}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should be rejected (empty title) + assert res.status_code == 400 + + # Test 2: XSS in tag title + res = client.post( + url_for("tag"), + data=json.dumps({"title": ""}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept but sanitize + if res.status_code == 201: + tag_uuid = res.json.get('uuid') + # Verify title is stored safely + res = client.get(url_for("tag", uuid=tag_uuid), headers={'x-api-key': api_key}) + # Should be escaped or sanitized + client.delete(url_for("tag", uuid=tag_uuid), headers={'x-api-key': api_key}) + + # Test 3: Path traversal in tag title + res = client.post( + url_for("tag"), + data=json.dumps({"title": "../../etc/passwd"}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should accept (it's just a string, not a path) + if res.status_code == 201: + tag_uuid = res.json.get('uuid') + client.delete(url_for("tag", uuid=tag_uuid), headers={'x-api-key': api_key}) + + # Test 4: Very long tag title + res = client.post( + url_for("tag"), + data=json.dumps({"title": "x" * 10000}), + headers={'content-type': 'application/json', 'x-api-key': api_key}, + ) + # Should be rejected (exceeds maxLength) + assert res.status_code == 400 + + +def test_api_authentication_edge_cases(client, live_server, measure_memory_usage, datastore_path): + """ + Test API authentication edge cases. + """ + api_key = live_server.app.config['DATASTORE'].data['settings']['application'].get('api_access_token') + set_original_response(datastore_path=datastore_path) + test_url = url_for('test_endpoint', _external=True) + + # Test 1: Missing API key + res = client.get(url_for("createwatch")) + assert res.status_code == 403, "Missing API key should be forbidden" + + # Test 2: Invalid API key + res = client.get( + url_for("createwatch"), + headers={'x-api-key': "invalid_key_12345"} + ) + assert res.status_code == 403, "Invalid API key should be forbidden" + + # Test 3: API key with special characters + res = client.get( + url_for("createwatch"), + headers={'x-api-key': "key"} + ) + assert res.status_code == 403, "Invalid API key should be forbidden" + + # Test 4: Very long API key + res = client.get( + url_for("createwatch"), + headers={'x-api-key': "x" * 10000} + ) + assert res.status_code == 403, "Invalid API key should be forbidden" + + # Test 5: Case sensitivity of API key + wrong_case_key = api_key.upper() if api_key.islower() else api_key.lower() + res = client.get( + url_for("createwatch"), + headers={'x-api-key': wrong_case_key} + ) + # Should be forbidden (keys are case-sensitive) + assert res.status_code == 403, "Wrong case API key should be forbidden" + + # Test 6: Valid API key should work + res = client.get( + url_for("createwatch"), + headers={'x-api-key': api_key} + ) + assert res.status_code == 200, "Valid API key should work" diff --git a/changedetectionio/validate_url.py b/changedetectionio/validate_url.py index 83ab592e4..1cb208d2c 100644 --- a/changedetectionio/validate_url.py +++ b/changedetectionio/validate_url.py @@ -64,6 +64,19 @@ def is_safe_valid_url(test_url): import re import validators + # Validate input type first - must be a non-empty string + if test_url is None: + logger.warning('URL validation failed: URL is None') + return False + + if not isinstance(test_url, str): + logger.warning(f'URL validation failed: URL must be a string, got {type(test_url).__name__}') + return False + + if not test_url.strip(): + logger.warning('URL validation failed: URL is empty or whitespace only') + return False + allow_file_access = strtobool(os.getenv('ALLOW_FILE_URI', 'false')) safe_protocol_regex = '^(http|https|ftp|file):' if allow_file_access else '^(http|https|ftp):' diff --git a/docs/api-spec.yaml b/docs/api-spec.yaml index 15eb252ff..d9d87aa4e 100644 --- a/docs/api-spec.yaml +++ b/docs/api-spec.yaml @@ -183,15 +183,30 @@ components: properties: weeks: type: integer + minimum: 0 + maximum: 52000 + nullable: true days: type: integer + minimum: 0 + maximum: 365000 + nullable: true hours: type: integer + minimum: 0 + maximum: 8760000 + nullable: true minutes: type: integer + minimum: 0 + maximum: 525600000 + nullable: true seconds: type: integer - description: Time intervals between checks + minimum: 0 + maximum: 31536000000 + nullable: true + description: Time intervals between checks. All fields must be non-negative. At least one non-zero value required when not using default settings. time_between_check_use_default: type: boolean default: true @@ -200,7 +215,9 @@ components: type: array items: type: string - description: Notification URLs for this web page change monitor (watch) + maxLength: 1000 + maxItems: 100 + description: Notification URLs for this web page change monitor (watch). Maximum 100 URLs. notification_title: type: string description: Custom notification title @@ -224,14 +241,19 @@ components: operation: type: string maxLength: 5000 + nullable: true selector: type: string maxLength: 5000 + nullable: true optional_value: type: string maxLength: 5000 + nullable: true required: [operation, selector, optional_value] - description: Browser automation steps + additionalProperties: false + maxItems: 100 + description: Browser automation steps. Maximum 100 steps allowed. processor: type: string enum: [restock_diff, text_json_diff]