feat: add validation for internal URLs in conditions_test

This commit is contained in:
arabcoders 2026-04-18 20:39:20 +03:00
parent 3f6149aaf7
commit 00ce7efb32
2 changed files with 47 additions and 0 deletions

View file

@ -15,6 +15,7 @@ from app.library.config import Config
from app.library.encoder import Encoder
from app.library.Events import EventBus, Events
from app.library.router import route
from app.library.Utils import validate_url
LOG: logging.Logger = logging.getLogger(__name__)
@ -83,6 +84,11 @@ async def conditions_test(request: Request, encoder: Encoder, cache: Cache, conf
if not (cond := params.get("condition")):
return web.json_response({"error": "condition is required."}, status=web.HTTPBadRequest.status_code)
try:
validate_url(url, allow_internal=config.allow_internal_urls)
except ValueError as e:
return web.json_response({"error": str(e)}, status=web.HTTPBadRequest.status_code)
try:
preset: str = params.get("preset", config.default_preset)
key: str = cache.hash(url + str(preset))

View file

@ -4,7 +4,14 @@ from __future__ import annotations
import pytest
import pytest_asyncio
from types import SimpleNamespace
import pytest
from aiohttp import web
from aiohttp.test_utils import make_mocked_request
from app.features.conditions.router import conditions_test
from app.library.config import Config
from app.library.encoder import Encoder
from app.features.conditions.repository import ConditionsRepository
from app.library.sqlite_store import SqliteStore
@ -33,6 +40,40 @@ async def repo():
SqliteStore._reset_singleton()
def _json_request(path: str, payload: object) -> web.Request:
request = make_mocked_request("POST", path)
async def _json() -> object:
return payload
request.json = _json # type: ignore[attr-defined]
return request
class TestAllowInternalUrlsScope:
def setup_method(self) -> None:
Config._reset_singleton()
@pytest.mark.asyncio
async def test_conditions_test_rejects_internal_url_when_disallowed(self) -> None:
config = Config.get_instance()
config.allow_internal_urls = False
encoder = Encoder()
cache = SimpleNamespace(hash=lambda value: value, has=lambda _key: False, set=lambda **_kwargs: None)
request = _json_request(
"/api/conditions/test/",
{"url": "http://127.0.0.1/test", "condition": "title", "preset": config.default_preset},
)
response = await conditions_test(request, encoder, cache, config)
assert response.status == web.HTTPBadRequest.status_code
body = response.body
assert isinstance(body, bytes)
assert b"internal urls" in body.lower()
class TestConditionsRepository:
"""Test suite for ConditionsRepository database operations."""