bitfennec
1059be1b62
- Rust core engine with HTTP ingestion gateway, hybrid search, and white-box resume matching - Python edge collectors supporting declarative TOML and specialized adapters - PostgreSQL schema migrations with pgvector embedding support - Hardened security controls: strict CSP, terminal sanitization, and input boundary validation - Multi-service container orchestration with dedicated non-root execution
122 lines
5.0 KiB
Python
122 lines
5.0 KiB
Python
import sys
|
|
from pathlib import Path
|
|
import httpx
|
|
|
|
# Ensure collectors directory is in python search path
|
|
current_dir = Path(__file__).resolve().parent
|
|
collector_root = current_dir.parent
|
|
if str(collector_root) not in sys.path:
|
|
sys.path.insert(0, str(collector_root))
|
|
|
|
from common.content import (
|
|
is_anti_crawler,
|
|
is_attachment_name,
|
|
sanitize_body_text,
|
|
)
|
|
from common.fetcher import PoliteFetcher
|
|
from common.security import is_safe_url
|
|
|
|
|
|
def test_is_safe_url():
|
|
# Valid urls
|
|
assert is_safe_url("https://www.yxrc.cn/detail", ["yxrc.cn"])
|
|
assert is_safe_url("https://sub.yxrc.cn/detail", ["yxrc.cn"])
|
|
assert is_safe_url("http://hrss.yn.gov.cn/article", ["hrss.yn.gov.cn"])
|
|
|
|
# Disallowed domains
|
|
assert not is_safe_url("https://evil.com/detail", ["yxrc.cn"])
|
|
assert not is_safe_url("https://attacker-yxrc.cn/detail", ["yxrc.cn"])
|
|
|
|
# Localhost and loopback IPv4
|
|
assert not is_safe_url("http://localhost:8080/ingest", ["localhost"])
|
|
assert not is_safe_url("http://127.0.0.1:8080/ingest")
|
|
assert not is_safe_url("http://127.0.1.1/ingest")
|
|
|
|
# Private IPv4 ranges
|
|
assert not is_safe_url("http://10.0.0.1/")
|
|
assert not is_safe_url("http://172.16.0.1/")
|
|
assert not is_safe_url("http://192.168.1.1/")
|
|
assert not is_safe_url("http://169.254.169.254/latest/meta-data")
|
|
|
|
# IPv6 literals and loopback
|
|
assert not is_safe_url("http://[::1]:8080/x")
|
|
assert not is_safe_url("http://[fe80::1]/x")
|
|
|
|
# IPv4-mapped IPv6
|
|
assert not is_safe_url("http://[::ffff:127.0.0.1]/x")
|
|
assert not is_safe_url("http://[::ffff:10.0.0.1]/x")
|
|
|
|
# Userinfo confusion
|
|
assert not is_safe_url("http://[email protected]/x")
|
|
|
|
|
|
def test_content_sanitization():
|
|
# Standalone attachment filename (< 120 chars) should be detected
|
|
assert is_attachment_name("2026年下半年玉溪市事业单位公开招聘工作人员岗位表.xlsx")
|
|
assert is_attachment_name("附件1：报考指南.pdf")
|
|
assert is_attachment_name("关于延期报名的补充说明.docx")
|
|
|
|
# Long text ending with attachment name should NOT be detected as attachment name (Regression N-1)
|
|
long_article = "招聘公告正文内容很多很多..." * 20 + "详情请参阅玉溪市拟聘公示(2022年3月28日）.docx"
|
|
assert len(long_article) > 120
|
|
assert not is_attachment_name(long_article)
|
|
assert sanitize_body_text(long_article) == long_article
|
|
|
|
# Short texts (< 60 chars) should be sanitized to None (N-3)
|
|
short_text = "国刑法修正案（九）》《治安管理处罚法》"
|
|
assert len(short_text) < 60
|
|
assert sanitize_body_text(short_text) is None
|
|
|
|
# Anti-crawler pages should be sanitized to None
|
|
assert is_anti_crawler("访问过于频繁，请完成验证后再试。")
|
|
assert sanitize_body_text("当前网络环境异常，请完成滑动验证后继续浏览本网站。") is None
|
|
|
|
# Long text mentioning security verification in context should NOT be treated as anti-crawler
|
|
long_notice_with_keyword = "报名流程注意事项：考生在进入报名系统前需完成手机短信安全验证方可继续提交报考材料。有关本次招聘的具体岗位、报考条件详见附件..." * 5
|
|
assert len(long_notice_with_keyword) > 300
|
|
assert not is_anti_crawler(long_notice_with_keyword)
|
|
assert sanitize_body_text(long_notice_with_keyword) == long_notice_with_keyword
|
|
|
|
# Valid real body text
|
|
valid_body = "根据《事业单位人事管理条例》及有关规定，经市委、市政府研究同意，现将2026年事业单位公开招聘工作人员有关事项公告如下：第一条 招聘岗位及人数..." * 3
|
|
assert sanitize_body_text(valid_body) == valid_body
|
|
|
|
|
|
def test_redirect_ssrf_blocked():
|
|
def mock_handler(request: httpx.Request):
|
|
if request.url.path == "/redirect-to-internal":
|
|
return httpx.Response(302, headers={"Location": "http://127.0.0.1:8080/secret"})
|
|
elif request.url.path == "/redirect-to-safe":
|
|
return httpx.Response(302, headers={"Location": "https://www.yxrc.cn/final"})
|
|
elif request.url.path == "/final":
|
|
return httpx.Response(200, text="Safe destination reached")
|
|
return httpx.Response(404)
|
|
|
|
fetcher = PoliteFetcher()
|
|
fetcher.client = httpx.Client(
|
|
transport=httpx.MockTransport(mock_handler), follow_redirects=False
|
|
)
|
|
|
|
# 1. Redirect to loopback/internal should be blocked immediately
|
|
blocked = False
|
|
try:
|
|
fetcher.get_text(
|
|
"https://www.yxrc.cn/redirect-to-internal", allowed_domains=["yxrc.cn"]
|
|
)
|
|
except ValueError:
|
|
blocked = True
|
|
assert blocked, "Failed to block redirect to internal IP address"
|
|
|
|
# 2. Redirect to legitimate allowed destination should succeed
|
|
res = fetcher.get_text(
|
|
"https://www.yxrc.cn/redirect-to-safe", allowed_domains=["yxrc.cn"]
|
|
)
|
|
assert res == "Safe destination reached"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
test_is_safe_url()
|
|
test_content_sanitization()
|
|
test_redirect_ssrf_blocked()
|
|
print("All security and content tests passed!")
|