|
|
|
@@ -0,0 +1,319 @@
|
|
|
|
|
# tests/test_security_hardening.py — Regression tests for ROADMAP #84
|
|
|
|
|
# (BUG-021 → BUG-034): sanitizer, path isolation, password policy, regex
|
|
|
|
|
# safety, webhook SSRF, rate limiting, audit IP and share-page escaping.
|
|
|
|
|
|
|
|
|
|
import time
|
|
|
|
|
from pathlib import Path
|
|
|
|
|
|
|
|
|
|
import pytest
|
|
|
|
|
|
|
|
|
|
from backend.services.errors import ServiceError
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-021/022 — HTML sanitizer
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestSanitizer:
|
|
|
|
|
def test_strips_event_handlers(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<img src="x" onerror="alert(1)">')
|
|
|
|
|
assert "onerror" not in out
|
|
|
|
|
assert "alert" not in out
|
|
|
|
|
|
|
|
|
|
def test_drops_script_content(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html("<p>ok</p><script>alert(1)</script>tail")
|
|
|
|
|
assert "<script" not in out
|
|
|
|
|
assert "alert" not in out
|
|
|
|
|
assert "ok" in out
|
|
|
|
|
assert "tail" in out
|
|
|
|
|
|
|
|
|
|
def test_blocks_javascript_urls(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<a href="javascript:alert(1)">x</a>')
|
|
|
|
|
assert "javascript:" not in out
|
|
|
|
|
assert ">x</a>" in out
|
|
|
|
|
|
|
|
|
|
def test_blocks_obfuscated_javascript_url(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<a href="javascript:alert(1)">x</a>')
|
|
|
|
|
assert "javascript" not in out.lower()
|
|
|
|
|
|
|
|
|
|
def test_keeps_safe_markup(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<p class="a">hi</p><a href="/x" data-vault="v" data-path="p.md">y</a>')
|
|
|
|
|
assert '<p class="a">hi</p>' in out
|
|
|
|
|
assert 'data-vault="v"' in out
|
|
|
|
|
assert 'href="/x"' in out
|
|
|
|
|
|
|
|
|
|
def test_allows_image_data_uri_but_not_html(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
good = sanitize_html('<img src="data:image/png;base64,AAAA">')
|
|
|
|
|
assert "data:image/png" in good
|
|
|
|
|
bad = sanitize_html('<img src="data:text/html,<script>alert(1)</script>">')
|
|
|
|
|
assert "data:text/html" not in bad
|
|
|
|
|
|
|
|
|
|
def test_heading_ids_preserved(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<h2 id="intro">Intro</h2>')
|
|
|
|
|
assert 'id="intro"' in out
|
|
|
|
|
|
|
|
|
|
def test_drops_style_attribute(self):
|
|
|
|
|
from backend.services.sanitizer import sanitize_html
|
|
|
|
|
out = sanitize_html('<p style="position:fixed">x</p>')
|
|
|
|
|
assert "style" not in out
|
|
|
|
|
|
|
|
|
|
def test_is_safe_url(self):
|
|
|
|
|
from backend.services.sanitizer import is_safe_url
|
|
|
|
|
assert is_safe_url("/relative")
|
|
|
|
|
assert is_safe_url("https://example.com")
|
|
|
|
|
assert is_safe_url("#anchor")
|
|
|
|
|
assert not is_safe_url("javascript:alert(1)")
|
|
|
|
|
assert not is_safe_url("vbscript:msgbox")
|
|
|
|
|
assert not is_safe_url("data:text/html,<script>")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-024 — Vault path isolation
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestPathIsolation:
|
|
|
|
|
def test_inside_vault_ok(self, tmp_path):
|
|
|
|
|
from backend.services.paths import resolve_safe_path
|
|
|
|
|
root = tmp_path / "vault"
|
|
|
|
|
root.mkdir()
|
|
|
|
|
resolved = resolve_safe_path(root, "note.md")
|
|
|
|
|
assert resolved == (root / "note.md").resolve()
|
|
|
|
|
|
|
|
|
|
def test_sibling_prefix_is_rejected(self, tmp_path):
|
|
|
|
|
from backend.services.paths import resolve_safe_path
|
|
|
|
|
(tmp_path / "vault").mkdir()
|
|
|
|
|
(tmp_path / "vault-evil").mkdir()
|
|
|
|
|
with pytest.raises(ServiceError) as exc:
|
|
|
|
|
resolve_safe_path(tmp_path / "vault", "../vault-evil/secret.md")
|
|
|
|
|
assert exc.value.code == "path_outside_vault"
|
|
|
|
|
|
|
|
|
|
def test_parent_traversal_rejected(self, tmp_path):
|
|
|
|
|
from backend.services.paths import resolve_safe_path
|
|
|
|
|
root = tmp_path / "vault"
|
|
|
|
|
root.mkdir()
|
|
|
|
|
with pytest.raises(ServiceError) as exc:
|
|
|
|
|
resolve_safe_path(root, "../../etc/passwd")
|
|
|
|
|
assert exc.value.code == "path_outside_vault"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-028 — Password policy
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestPasswordPolicy:
|
|
|
|
|
def test_short_rejected(self):
|
|
|
|
|
from backend.auth.password import validate_password_strength
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_password_strength("short")
|
|
|
|
|
|
|
|
|
|
def test_too_long_rejected(self):
|
|
|
|
|
from backend.auth.password import validate_password_strength
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_password_strength("a" * 200)
|
|
|
|
|
|
|
|
|
|
def test_valid_accepted(self):
|
|
|
|
|
from backend.auth.password import validate_password_strength
|
|
|
|
|
assert validate_password_strength("ValidPass123") == "ValidPass123"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-025 — Regex safety (ReDoS)
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestRegexSafety:
|
|
|
|
|
def test_nested_quantifier_rejected(self):
|
|
|
|
|
from backend.services.regex_safety import validate_regex
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_regex("(a+)+$")
|
|
|
|
|
|
|
|
|
|
def test_long_pattern_rejected(self):
|
|
|
|
|
from backend.services.regex_safety import validate_regex
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_regex("a" * 600)
|
|
|
|
|
|
|
|
|
|
def test_simple_pattern_accepted(self):
|
|
|
|
|
from backend.services.regex_safety import validate_regex
|
|
|
|
|
assert validate_regex(r"\bpython\b") == r"\bpython\b"
|
|
|
|
|
|
|
|
|
|
def test_truncate_for_regex(self):
|
|
|
|
|
from backend.services.regex_safety import truncate_for_regex
|
|
|
|
|
assert len(truncate_for_regex("x" * 1000, limit=10)) == 10
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-026 — Webhook SSRF
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestWebhookUrlValidation:
|
|
|
|
|
def test_https_public_ok(self):
|
|
|
|
|
from backend.webhooks import validate_webhook_url
|
|
|
|
|
assert validate_webhook_url("https://example.com/hook") == "https://example.com/hook"
|
|
|
|
|
|
|
|
|
|
def test_http_rejected_by_default(self, monkeypatch):
|
|
|
|
|
from backend.webhooks import validate_webhook_url
|
|
|
|
|
monkeypatch.delenv("OBSIGATE_WEBHOOK_ALLOW_HTTP", raising=False)
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_webhook_url("http://example.com/hook")
|
|
|
|
|
|
|
|
|
|
def test_private_ip_rejected(self, monkeypatch):
|
|
|
|
|
from backend.webhooks import validate_webhook_url
|
|
|
|
|
monkeypatch.delenv("OBSIGATE_WEBHOOK_ALLOW_PRIVATE", raising=False)
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_webhook_url("https://127.0.0.1/hook")
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_webhook_url("https://169.254.169.254/latest/meta-data")
|
|
|
|
|
|
|
|
|
|
def test_bad_scheme_rejected(self):
|
|
|
|
|
from backend.webhooks import validate_webhook_url
|
|
|
|
|
with pytest.raises(ValueError):
|
|
|
|
|
validate_webhook_url("ftp://example.com")
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-031 — Per-account rate limiting
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestAccountRateLimit:
|
|
|
|
|
def test_account_limited_after_threshold(self, monkeypatch):
|
|
|
|
|
from backend import ratelimit
|
|
|
|
|
monkeypatch.setattr(ratelimit, "ACCOUNT_MAX_ATTEMPTS", 3)
|
|
|
|
|
account = "ratelimit-account@test"
|
|
|
|
|
assert not ratelimit.is_account_rate_limited(account)
|
|
|
|
|
for _ in range(3):
|
|
|
|
|
ratelimit.record_account_failure(account)
|
|
|
|
|
assert ratelimit.is_account_rate_limited(account)
|
|
|
|
|
ratelimit.record_account_success(account)
|
|
|
|
|
assert not ratelimit.is_account_rate_limited(account)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-030 — Client IP resolution
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestClientIp:
|
|
|
|
|
def _request(self, headers=None, client=("1.2.3.4", 1234)):
|
|
|
|
|
from starlette.requests import Request
|
|
|
|
|
raw_headers = [(k.lower().encode(), v.encode()) for k, v in (headers or {}).items()]
|
|
|
|
|
scope = {
|
|
|
|
|
"type": "http",
|
|
|
|
|
"method": "GET",
|
|
|
|
|
"path": "/",
|
|
|
|
|
"query_string": b"",
|
|
|
|
|
"headers": raw_headers,
|
|
|
|
|
"client": client,
|
|
|
|
|
"scheme": "http",
|
|
|
|
|
"server": ("test", 80),
|
|
|
|
|
}
|
|
|
|
|
return Request(scope)
|
|
|
|
|
|
|
|
|
|
def test_socket_peer_used_by_default(self, monkeypatch):
|
|
|
|
|
from backend.services.net import get_client_ip
|
|
|
|
|
monkeypatch.delenv("OBSIGATE_TRUST_PROXY", raising=False)
|
|
|
|
|
assert get_client_ip(self._request()) == "1.2.3.4"
|
|
|
|
|
|
|
|
|
|
def test_forwarded_header_used_when_trusted(self, monkeypatch):
|
|
|
|
|
from backend.services.net import get_client_ip
|
|
|
|
|
monkeypatch.setenv("OBSIGATE_TRUST_PROXY", "true")
|
|
|
|
|
req = self._request({"X-Forwarded-For": "9.8.7.6, 10.0.0.1"})
|
|
|
|
|
assert get_client_ip(req) == "9.8.7.6"
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-032 — Indexing must not follow symlinks outside the vault
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestSymlinkIndexing:
|
|
|
|
|
def test_external_symlink_is_skipped(self, tmp_path):
|
|
|
|
|
import os
|
|
|
|
|
|
|
|
|
|
from backend.indexer import _scan_vault
|
|
|
|
|
|
|
|
|
|
vault = tmp_path / "vault"
|
|
|
|
|
vault.mkdir()
|
|
|
|
|
(vault / "inside.md").write_text("# Inside\n", encoding="utf-8")
|
|
|
|
|
|
|
|
|
|
outside = tmp_path / "outside"
|
|
|
|
|
outside.mkdir()
|
|
|
|
|
(outside / "secret.md").write_text("# Secret\n", encoding="utf-8")
|
|
|
|
|
|
|
|
|
|
link = vault / "link.md"
|
|
|
|
|
try:
|
|
|
|
|
os.symlink(outside / "secret.md", link)
|
|
|
|
|
except (OSError, NotImplementedError):
|
|
|
|
|
pytest.skip("Symlinks are not supported in this environment")
|
|
|
|
|
|
|
|
|
|
result = _scan_vault("vault", str(vault))
|
|
|
|
|
paths = {f["path"] for f in result["files"]}
|
|
|
|
|
assert "inside.md" in paths
|
|
|
|
|
assert "link.md" not in paths
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-021/022 — End-to-end rendering & share page escaping
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestRenderingAndShare:
|
|
|
|
|
def test_markdown_rendering_is_sanitized(self, client, test_vault_dir):
|
|
|
|
|
payload = (
|
|
|
|
|
"---\ntitle: Safe\n---\n"
|
|
|
|
|
"# Hello\n\n"
|
|
|
|
|
"<img src=x onerror=alert(1)>\n\n"
|
|
|
|
|
"<script>alert(2)</script>\n\n"
|
|
|
|
|
"Normal **text**.\n"
|
|
|
|
|
)
|
|
|
|
|
(Path(test_vault_dir) / "xss.md").write_text(payload, encoding="utf-8")
|
|
|
|
|
resp = client.get("/api/file/TestVault", params={"path": "xss.md"})
|
|
|
|
|
assert resp.status_code == 200
|
|
|
|
|
html = resp.json()["html"]
|
|
|
|
|
assert "onerror" not in html
|
|
|
|
|
assert "<script" not in html
|
|
|
|
|
assert "<strong>text</strong>" in html
|
|
|
|
|
|
|
|
|
|
def test_share_page_escapes_title_and_frontmatter(self, client, test_vault_dir):
|
|
|
|
|
payload = (
|
|
|
|
|
'---\ntitle: "<script>alert(1)</script>"\n'
|
|
|
|
|
'author: "<img src=x onerror=alert(9)>"\n---\n'
|
|
|
|
|
"# Body\ncontent\n"
|
|
|
|
|
)
|
|
|
|
|
(Path(test_vault_dir) / "share_xss.md").write_text(payload, encoding="utf-8")
|
|
|
|
|
create = client.post("/api/share/TestVault", json={"path": "share_xss.md"})
|
|
|
|
|
assert create.status_code == 200
|
|
|
|
|
token = create.json()["token"]
|
|
|
|
|
|
|
|
|
|
page = client.get(f"/s/{token}")
|
|
|
|
|
assert page.status_code == 200
|
|
|
|
|
body = page.text
|
|
|
|
|
assert "<script>alert(1)</script>" not in body
|
|
|
|
|
assert "<script>alert(1)</script>" in body
|
|
|
|
|
# The img tag must be escaped, so no live attribute is emitted.
|
|
|
|
|
assert "<img src=x onerror" not in body
|
|
|
|
|
assert "<img" in body
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
# BUG-027/028 — Token invalidation on password change
|
|
|
|
|
# ═══════════════════════════════════════════════════════════════════
|
|
|
|
|
|
|
|
|
|
class TestTokenInvalidation:
|
|
|
|
|
def test_access_token_rejected_after_password_change(self, admin_client):
|
|
|
|
|
login = admin_client.post(
|
|
|
|
|
"/api/auth/login", json={"username": "admin", "password": "chab30"}
|
|
|
|
|
)
|
|
|
|
|
assert login.status_code == 200
|
|
|
|
|
token = login.json()["access_token"]
|
|
|
|
|
|
|
|
|
|
# Simulate a password change happening in the future relative to the token.
|
|
|
|
|
from backend.auth.user_store import update_user
|
|
|
|
|
update_user("admin", {"password_changed_at": time.time() + 10})
|
|
|
|
|
|
|
|
|
|
me = admin_client.get(
|
|
|
|
|
"/api/auth/me", headers={"Authorization": f"Bearer {token}"}
|
|
|
|
|
)
|
|
|
|
|
assert me.status_code == 401
|