edu-industry/admin-tools/crawler/tests/test_security.py

from dataclasses import replace
import os

import httpx
import pytest

from edu_crawler.errors import (
    AuthOrChallenge,
    ConfigurationError,
    HostNotAllowed,
    PrivateAddressBlocked,
    RobotsDenied,
    SourceStructureChanged,
)
from edu_crawler.http import SafeHttpClient
from edu_crawler.policies import source_policy
from edu_crawler.security import (
    prohibited_school_keys,
    prohibited_school_values,
    redact_url,
    safe_record_reference,
    validate_public_url,
)
from edu_crawler.transport import PinnedNetworkBackend


def public_resolver(_host: str, _port: int, **_kwargs: object) -> list[tuple]:
    return [(2, 1, 6, "", ("93.184.216.34", 443))]


def private_resolver(_host: str, _port: int, **_kwargs: object) -> list[tuple]:
    return [(2, 1, 6, "", ("127.0.0.1", 443))]


def test_url_allowlist_and_ssrf_controls() -> None:
    assert (
        validate_public_url(
            "https://brzp.chinasme.cn/jobs?page=1",
            ("brzp.chinasme.cn",),
            resolver=public_resolver,
        )
        == "https://brzp.chinasme.cn/jobs?page=1"
    )
    with pytest.raises(HostNotAllowed):
        validate_public_url(
            "https://evil.example/jobs",
            ("brzp.chinasme.cn",),
            resolver=public_resolver,
        )
    credential_url = "https://" + "user" + ":" + "example" + "@brzp.chinasme.cn/jobs"
    with pytest.raises(HostNotAllowed):
        validate_public_url(
            credential_url, ("brzp.chinasme.cn",), resolver=public_resolver
        )
    with pytest.raises(PrivateAddressBlocked):
        validate_public_url(
            "https://brzp.chinasme.cn/jobs",
            ("brzp.chinasme.cn",),
            resolver=private_resolver,
        )
    with pytest.raises(PrivateAddressBlocked):
        validate_public_url(
            "https://127.0.0.1/jobs", ("127.0.0.1",), resolver=public_resolver
        )


def test_url_redaction_removes_credentials_and_fragment() -> None:
    credential_url = "https://" + "user" + ":" + "example" + "@example.com"
    value = redact_url(credential_url + "/jobs?token=abc&city=330100#private")
    assert value == "https://example.com/jobs?token=%5BREDACTED%5D&city=330100"
    assert "pass" not in value
    assert "abc" not in value
    assert "private" not in value


def test_record_reference_is_sanitized_before_quarantine_or_ids() -> None:
    assert safe_record_reference("contact student@example.com") is None
    assert (
        safe_record_reference(
            {"value": "https://example.com/job/1?token=secret#private"}
        )
        == "https://example.com/job/1?token=%5BREDACTED%5D"
    )


def test_nested_school_pii_detection() -> None:
    record = {
        "student_token": "STU_x",
        "profile": {"student_name": "不应进入平台"},
        "student": {"name": "不应进入平台"},
        "rows": [{"phone": "1"}],
    }
    assert prohibited_school_keys(record) == [
        "profile.student_name",
        "rows[0].phone",
        "student.name",
    ]


def test_school_pii_value_detection_reports_paths_only() -> None:
    record = {
        "student_token": "STU_" + "a" * 32,
        "source_record_id": "EMP-001",
        "comment": "联系 13800138000 或 student@example.com",
    }
    assert prohibited_school_values(record) == ["comment"]


def test_http_client_rejects_unversioned_approved_host() -> None:
    policy = replace(source_policy("BRZP"), minimum_interval_seconds=0)
    with pytest.raises(ConfigurationError):
        SafeHttpClient(
            policy,
            contact="crawler@example.edu.cn",
            approved_hosts=("new-host.example",),
            resolver=public_resolver,
            transport=httpx.MockTransport(lambda _request: httpx.Response(200)),
        )


def test_robots_denial_stops_before_target_request() -> None:
    requests: list[str] = []

    def handler(request: httpx.Request) -> httpx.Response:
        requests.append(str(request.url))
        assert request.url.path == "/robots.txt"
        return httpx.Response(
            200,
            headers={"content-type": "text/plain"},
            text="User-agent: *\nDisallow: /private",
        )

    policy = replace(source_policy("BRZP"), minimum_interval_seconds=0)
    with SafeHttpClient(
        policy,
        contact="crawler@example.edu.cn",
        approved_hosts=("brzp.chinasme.cn",),
        resolver=public_resolver,
        transport=httpx.MockTransport(handler),
    ) as client:
        with pytest.raises(RobotsDenied):
            client.fetch("https://brzp.chinasme.cn/private")
    assert requests == ["https://brzp.chinasme.cn/robots.txt"]


def test_challenge_page_stops_without_bypass() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/robots.txt":
            return httpx.Response(
                200,
                headers={"content-type": "text/plain"},
                text="User-agent: *\nAllow: /",
            )
        return httpx.Response(
            200, headers={"content-type": "text/html"}, text="<title>安全验证</title>"
        )

    policy = replace(source_policy("BRZP"), minimum_interval_seconds=0)
    with SafeHttpClient(
        policy,
        contact="crawler@example.edu.cn",
        approved_hosts=("brzp.chinasme.cn",),
        resolver=public_resolver,
        transport=httpx.MockTransport(handler),
    ) as client:
        with pytest.raises(AuthOrChallenge):
            client.fetch("https://brzp.chinasme.cn/job/1")


def test_network_errors_do_not_echo_sensitive_urls() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        raise httpx.ConnectError("failed token=secret-value", request=request)

    policy = replace(
        source_policy("BRZP"), minimum_interval_seconds=0, robots_required=False
    )
    with SafeHttpClient(
        policy,
        contact="crawler@example.edu.cn",
        approved_hosts=("brzp.chinasme.cn",),
        resolver=public_resolver,
        transport=httpx.MockTransport(handler),
    ) as client:
        with pytest.raises(SourceStructureChanged) as caught:
            client.fetch("https://brzp.chinasme.cn/jobs?token=secret-value")
    assert str(caught.value) == "Source network request failed"


def test_redirect_is_revalidated_against_allowlist() -> None:
    def handler(request: httpx.Request) -> httpx.Response:
        if request.url.path == "/robots.txt":
            return httpx.Response(
                200,
                headers={"content-type": "text/plain"},
                text="User-agent: *\nAllow: /",
            )
        return httpx.Response(302, headers={"location": "https://evil.example/private"})

    policy = replace(source_policy("BRZP"), minimum_interval_seconds=0)
    with SafeHttpClient(
        policy,
        contact="crawler@example.edu.cn",
        approved_hosts=("brzp.chinasme.cn",),
        resolver=public_resolver,
        transport=httpx.MockTransport(handler),
    ) as client:
        with pytest.raises(HostNotAllowed):
            client.fetch("https://brzp.chinasme.cn/job/1")


def test_bearer_token_is_not_sent_to_robots(monkeypatch: pytest.MonkeyPatch) -> None:
    observed: list[tuple[str, str | None]] = []

    def handler(request: httpx.Request) -> httpx.Response:
        observed.append((request.url.path, request.headers.get("authorization")))
        if request.url.path == "/robots.txt":
            return httpx.Response(
                200,
                headers={"content-type": "text/plain"},
                text="User-agent: *\nAllow: /",
            )
        return httpx.Response(
            200, headers={"content-type": "application/json"}, json=[{"id": "one"}]
        )

    monkeypatch.setenv("TEST_SOURCE_BEARER", "example-token-value")
    policy = replace(source_policy("BRZP"), minimum_interval_seconds=0)
    with SafeHttpClient(
        policy,
        contact="crawler@example.edu.cn",
        approved_hosts=("brzp.chinasme.cn",),
        bearer_token_env="TEST_SOURCE_BEARER",
        resolver=public_resolver,
        transport=httpx.MockTransport(handler),
    ) as client:
        client.fetch("https://brzp.chinasme.cn/jobs")
    assert observed == [
        ("/robots.txt", None),
        ("/jobs", "Bearer example-token-value"),
    ]
    assert os.environ["TEST_SOURCE_BEARER"] == "example-token-value"


def test_pinned_backend_connects_to_the_validated_ip(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    connected: list[tuple] = []

    class FakeSocket:
        def settimeout(self, _timeout: float | None) -> None:
            pass

        def setsockopt(self, *_option: object) -> None:
            pass

        def bind(self, _address: tuple) -> None:
            pass

        def connect(self, address: tuple) -> None:
            connected.append(address)

        def close(self) -> None:
            pass

    monkeypatch.setattr(
        "edu_crawler.transport.socket.socket", lambda *_args: FakeSocket()
    )
    stream = PinnedNetworkBackend(public_resolver).connect_tcp("brzp.chinasme.cn", 443)
    stream.close()
    assert connected == [("93.184.216.34", 443)]