153 lines
4.8 KiB
Python
153 lines
4.8 KiB
Python
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
from django.utils import timezone
|
|
|
|
from apps.sources.models import Source, SourceRobotsCache
|
|
from apps.sources.services.robots import assess_robots
|
|
from apps.sources.services.url_security import ValidatedUrl
|
|
|
|
|
|
def _allow_test_dns(monkeypatch):
|
|
monkeypatch.setattr(
|
|
"apps.sources.services.robots.validate_public_url",
|
|
lambda url, **kwargs: ValidatedUrl(
|
|
url=url,
|
|
hostname="jobs.example.org",
|
|
port=443,
|
|
addresses=("93.184.216.34",),
|
|
),
|
|
)
|
|
|
|
|
|
@pytest.mark.security
|
|
def test_robots_uses_user_agent_matching_and_allows_default(db, monkeypatch):
|
|
_allow_test_dns(monkeypatch)
|
|
source = Source.objects.create(
|
|
name="Example jobs",
|
|
source_type=Source.Type.EMPLOYER,
|
|
base_url="https://jobs.example.org/vacatures/",
|
|
domain="jobs.example.org",
|
|
status=Source.Status.ACTIVE,
|
|
policy=Source.Policy.ALLOW,
|
|
)
|
|
calls = {"count": 0}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
calls["count"] += 1
|
|
assert request.url.path == "/robots.txt"
|
|
body = (
|
|
"User-agent: *\n"
|
|
"Disallow: /admin\n"
|
|
"User-agent: vacatureradar/0.1 (+local-personal-use)\n"
|
|
"Allow: /admin\n"
|
|
)
|
|
return httpx.Response(200, text=body)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
decision = assess_robots(
|
|
source.base_url + "admin",
|
|
source=source,
|
|
user_agent="VacatureRadar/0.1 (+local-personal-use)",
|
|
client=client,
|
|
now=timezone.now(),
|
|
)
|
|
client.close()
|
|
assert decision.allowed
|
|
assert calls["count"] == 1
|
|
|
|
|
|
@pytest.mark.security
|
|
def test_robots_cache_refreshes_after_ttl(db, settings, monkeypatch):
|
|
_allow_test_dns(monkeypatch)
|
|
source = Source.objects.create(
|
|
name="Example jobs",
|
|
source_type=Source.Type.EMPLOYER,
|
|
base_url="https://jobs.example.org/vacatures/",
|
|
domain="jobs.example.org",
|
|
status=Source.Status.ACTIVE,
|
|
policy=Source.Policy.ALLOW,
|
|
)
|
|
settings.ROBOTS_CACHE_TTL_SECONDS = 0
|
|
|
|
calls = {"count": 0}
|
|
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
calls["count"] += 1
|
|
return httpx.Response(200, text="User-agent: *\nAllow: /")
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(handler))
|
|
assess_robots(
|
|
source.base_url, source=source, user_agent="test-agent", client=client, now=timezone.now()
|
|
)
|
|
assert calls["count"] == 1
|
|
assess_robots(
|
|
source.base_url,
|
|
source=source,
|
|
user_agent="test-agent",
|
|
client=client,
|
|
now=timezone.now(),
|
|
)
|
|
assert calls["count"] == 2
|
|
assert SourceRobotsCache.objects.filter(origin__contains="jobs.example.org").exists()
|
|
client.close()
|
|
|
|
|
|
@pytest.mark.security
|
|
def test_robots_follows_and_revalidates_redirects(db, monkeypatch):
|
|
validated_urls = []
|
|
|
|
def validate(url, **kwargs):
|
|
validated_urls.append(url)
|
|
if "127.0.0.1" in url:
|
|
from apps.sources.services.url_security import UnsafeUrlError
|
|
|
|
raise UnsafeUrlError("Privéadres")
|
|
return ValidatedUrl(
|
|
url=url,
|
|
hostname="jobs.example.org",
|
|
port=443,
|
|
addresses=("93.184.216.34",),
|
|
)
|
|
|
|
monkeypatch.setattr("apps.sources.services.robots.validate_public_url", validate)
|
|
source = Source.objects.create(
|
|
name="Redirect jobs",
|
|
source_type=Source.Type.ATS,
|
|
base_url="https://ats.example.org/jobs",
|
|
domain="ats.example.org",
|
|
status=Source.Status.ACTIVE,
|
|
policy=Source.Policy.ALLOW,
|
|
)
|
|
|
|
def safe_handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.host == "ats.example.org":
|
|
return httpx.Response(
|
|
302,
|
|
headers={"location": "https://jobs.example.org/robots.txt"},
|
|
request=request,
|
|
)
|
|
return httpx.Response(200, text="User-agent: *\nAllow: /", request=request)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(safe_handler))
|
|
assert assess_robots(source.base_url, source=source, client=client).allowed is True
|
|
client.close()
|
|
assert validated_urls == [
|
|
"https://ats.example.org/robots.txt",
|
|
"https://jobs.example.org/robots.txt",
|
|
]
|
|
|
|
SourceRobotsCache.objects.all().delete()
|
|
|
|
def unsafe_handler(request: httpx.Request) -> httpx.Response:
|
|
return httpx.Response(
|
|
302, headers={"location": "http://127.0.0.1/robots.txt"}, request=request
|
|
)
|
|
|
|
client = httpx.Client(transport=httpx.MockTransport(unsafe_handler))
|
|
blocked = assess_robots(source.base_url, source=source, client=client)
|
|
client.close()
|
|
assert blocked.allowed is False
|
|
assert "Privéadres" in blocked.reason
|