317 lines
12 KiB
Python
317 lines
12 KiB
Python
"""
|
|
Offensive security tests — Phase 2.
|
|
|
|
Covers:
|
|
- Path traversal attacks (../../../etc/passwd, encoded variants, null bytes)
|
|
- Hostile filenames (null bytes, path separators, control chars)
|
|
- Symlink escape attempts via safe_rel_path/is_within_root
|
|
- Forged X-Forwarded-For / X-Forwarded-Host headers
|
|
- CSRF protection: missing anti-CSRF header, wrong Origin, null Origin
|
|
- Malformed metadata payloads (no crash / structured Err)
|
|
- Oversized inputs (no crash / truncation)
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock, patch
|
|
|
|
import pytest
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 1. Path traversal — safe_rel_path rejects all traversal attempts
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestPathTraversal:
|
|
"""
|
|
safe_rel_path() must return None for any input that could escape
|
|
the allowed root. Tests use OWASP-inspired payload list.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _import(self):
|
|
from mjr_am_backend import path_utils
|
|
self.safe_rel_path = path_utils.safe_rel_path
|
|
self.normalize_path = path_utils.normalize_path
|
|
|
|
# NOTE: URL percent-encoded payloads (%2e%2e%2f…) are intentionally excluded —
|
|
# URL decoding happens at the HTTP framework layer (aiohttp) BEFORE safe_rel_path
|
|
# is ever called. Testing them here would test the wrong layer.
|
|
# NOTE: four-dot sequences (....//...) are valid directory names, not traversal.
|
|
TRAVERSAL_PAYLOADS = [
|
|
"../../../etc/passwd",
|
|
"..\\..\\..\\windows\\system32\\drivers\\etc\\hosts",
|
|
"subfolder/../../etc/passwd",
|
|
"/etc/passwd",
|
|
"C:\\Windows\\System32",
|
|
"C:/Windows/System32",
|
|
"\x00evil", # null byte
|
|
"a/../../b",
|
|
"a\\..\\..\\b",
|
|
"../",
|
|
"..",
|
|
]
|
|
|
|
@pytest.mark.parametrize("payload", TRAVERSAL_PAYLOADS)
|
|
def test_safe_rel_path_rejects_traversal(self, payload: str):
|
|
result = self.safe_rel_path(payload)
|
|
assert result is None or result == Path(""), (
|
|
f"safe_rel_path({payload!r}) should return None or Path(''), got {result!r}"
|
|
)
|
|
|
|
def test_normalize_path_rejects_null_byte(self):
|
|
assert self.normalize_path("valid/path\x00injected") is None
|
|
|
|
def test_normalize_path_resolves_dotdot(self, tmp_path: Path):
|
|
"""normalize_path resolves .. but the result may be outside root — caller must check."""
|
|
result = self.normalize_path(str(tmp_path / "a" / ".." / ".."))
|
|
# Result is a valid absolute path, but callers MUST use is_within_root to guard it
|
|
assert result is not None
|
|
assert result.is_absolute()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 2. is_within_root — must block escape
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestIsWithinRoot:
|
|
@pytest.fixture(autouse=True)
|
|
def _import(self):
|
|
from mjr_am_backend import path_utils
|
|
self.is_within_root = path_utils.is_within_root
|
|
|
|
def test_path_within_root_is_allowed(self, tmp_path: Path):
|
|
subdir = tmp_path / "outputs" / "image.png"
|
|
subdir.parent.mkdir(parents=True, exist_ok=True)
|
|
subdir.touch()
|
|
assert self.is_within_root(subdir, tmp_path)
|
|
|
|
def test_path_outside_root_is_blocked(self, tmp_path: Path):
|
|
root = tmp_path / "outputs"
|
|
root.mkdir()
|
|
outside = tmp_path / "private" / "secret.txt"
|
|
outside.parent.mkdir()
|
|
outside.touch()
|
|
assert not self.is_within_root(outside, root)
|
|
|
|
def test_path_equal_to_root_is_allowed(self, tmp_path: Path):
|
|
assert self.is_within_root(tmp_path, tmp_path)
|
|
|
|
def test_nonexistent_path_does_not_raise(self, tmp_path: Path):
|
|
"""is_within_root must not raise if the path does not exist (strict=True returns False)."""
|
|
nonexistent = tmp_path / "ghost" / "file.png"
|
|
result = self.is_within_root(nonexistent, tmp_path)
|
|
# Strict mode: unresolvable path → False, never raises
|
|
assert isinstance(result, bool)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 3. Hostile filenames — safe_download_filename strips dangerous chars
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestHostileFilenames:
|
|
@pytest.fixture(autouse=True)
|
|
def _import(self):
|
|
from mjr_am_backend.routes.assets import path_guard
|
|
self.safe_download_filename = path_guard.safe_download_filename
|
|
|
|
HOSTILE_NAMES = [
|
|
('image "with" quotes.png', '"'),
|
|
('file\nwith\nnewlines.png', '\n'),
|
|
('file\rwith\rcr.png', '\r'),
|
|
('semi;colon.png', ';'),
|
|
('../traversal.png', '/'),
|
|
('..\\traversal.png', '\\'),
|
|
('C:drive.png', ':'),
|
|
('a' * 512 + '.png', None), # oversized
|
|
]
|
|
|
|
@pytest.mark.parametrize("name,banned_char", HOSTILE_NAMES)
|
|
def test_safe_download_filename_sanitizes(self, name: str, banned_char: str | None):
|
|
result = self.safe_download_filename(name)
|
|
assert isinstance(result, str)
|
|
if banned_char:
|
|
assert banned_char not in result, (
|
|
f"Banned char {banned_char!r} still present in {result!r}"
|
|
)
|
|
# Must not be empty
|
|
assert len(result) > 0
|
|
assert len(result) <= 255
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 4. CSRF protection tests
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestCSRFProtection:
|
|
@pytest.fixture(autouse=True)
|
|
def _import(self):
|
|
from mjr_am_backend.routes.core import security_csrf
|
|
self.csrf_error = security_csrf._csrf_error
|
|
|
|
def _make_request(self, method="POST", headers=None):
|
|
req = MagicMock()
|
|
req.method = method
|
|
req.headers = headers or {}
|
|
return req
|
|
|
|
def test_get_request_is_not_csrf_protected(self):
|
|
req = self._make_request(method="GET")
|
|
assert self.csrf_error(req) is None
|
|
|
|
def test_post_without_csrf_header_returns_error(self):
|
|
req = self._make_request(method="POST", headers={"Host": "localhost:8188"})
|
|
error = self.csrf_error(req)
|
|
assert error is not None
|
|
assert "CSRF" in error or "anti-CSRF" in error.lower() or "X-Requested-With" in error
|
|
|
|
def test_post_with_x_requested_with_passes(self):
|
|
req = self._make_request(
|
|
method="POST",
|
|
headers={"Host": "localhost:8188", "X-Requested-With": "XMLHttpRequest"},
|
|
)
|
|
assert self.csrf_error(req) is None
|
|
|
|
def test_post_with_csrf_token_header_passes(self):
|
|
req = self._make_request(
|
|
method="POST",
|
|
headers={"Host": "localhost:8188", "X-CSRF-Token": "some-value"},
|
|
)
|
|
assert self.csrf_error(req) is None
|
|
|
|
def test_null_origin_is_blocked(self):
|
|
req = self._make_request(
|
|
method="POST",
|
|
headers={
|
|
"Host": "localhost:8188",
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
"Origin": "null", # sandboxed iframe / file:// attacker
|
|
},
|
|
)
|
|
error = self.csrf_error(req)
|
|
assert error is not None
|
|
assert "null" in error.lower() or "blocked" in error.lower()
|
|
|
|
def test_cross_origin_is_blocked(self):
|
|
req = self._make_request(
|
|
method="POST",
|
|
headers={
|
|
"Host": "localhost:8188",
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
"Origin": "https://evil.com",
|
|
},
|
|
)
|
|
error = self.csrf_error(req)
|
|
assert error is not None
|
|
assert "evil.com" in error or "blocked" in error.lower()
|
|
|
|
def test_same_origin_loopback_passes(self):
|
|
req = self._make_request(
|
|
method="POST",
|
|
headers={
|
|
"Host": "localhost:8188",
|
|
"X-Requested-With": "XMLHttpRequest",
|
|
"Origin": "http://localhost:8188",
|
|
},
|
|
)
|
|
assert self.csrf_error(req) is None
|
|
|
|
def test_delete_without_csrf_header_is_blocked(self):
|
|
req = self._make_request(method="DELETE", headers={"Host": "localhost:8188"})
|
|
error = self.csrf_error(req)
|
|
assert error is not None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 5. Forged X-Forwarded-For / X-Forwarded-Host
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestForgedProxyHeaders:
|
|
@pytest.fixture(autouse=True)
|
|
def _import(self):
|
|
from mjr_am_backend.routes.core import security as sec
|
|
self.resolve_client_ip = sec._resolve_client_ip
|
|
self.is_loopback_ip = sec._is_loopback_ip
|
|
self.sec = sec
|
|
|
|
def test_xff_from_untrusted_proxy_is_ignored(self, monkeypatch):
|
|
"""X-Forwarded-For from a non-trusted proxy must not be trusted."""
|
|
from mjr_am_backend.routes.core import security as sec
|
|
|
|
# Peer is a non-trusted external IP
|
|
with patch.object(sec, "_is_trusted_proxy", return_value=False):
|
|
client_ip = self.resolve_client_ip("203.0.113.1", {"X-Forwarded-For": "8.8.8.8"})
|
|
# When peer is NOT trusted, XFF must be ignored
|
|
assert client_ip == "203.0.113.1"
|
|
|
|
def test_xff_from_trusted_proxy_is_respected(self, monkeypatch):
|
|
"""X-Forwarded-For from a trusted proxy must resolve to the claimed client IP."""
|
|
from mjr_am_backend.routes.core import security as sec
|
|
|
|
with patch.object(sec, "_is_trusted_proxy", lambda ip: ip in {"127.0.0.1"}):
|
|
client_ip = self.resolve_client_ip("127.0.0.1", {"X-Forwarded-For": "8.8.8.8, 127.0.0.1"})
|
|
assert client_ip == "8.8.8.8"
|
|
|
|
def test_forged_xff_claims_loopback_but_peer_is_external(self, monkeypatch):
|
|
"""
|
|
An attacker may forge X-Forwarded-For: 127.0.0.1 from an external IP.
|
|
The resolved client IP must NOT be loopback in that case.
|
|
"""
|
|
from mjr_am_backend.routes.core import security as sec
|
|
|
|
with patch.object(sec, "_is_trusted_proxy", return_value=False):
|
|
client_ip = self.resolve_client_ip("203.0.113.99", {"X-Forwarded-For": "127.0.0.1"})
|
|
# XFF is ignored because peer is not a trusted proxy
|
|
assert client_ip == "203.0.113.99"
|
|
assert not self.is_loopback_ip(client_ip)
|
|
|
|
def test_write_access_blocked_for_forged_loopback_xff(self, monkeypatch):
|
|
"""
|
|
Even if an attacker sends X-Forwarded-For: 127.0.0.1, write access
|
|
must not be granted when the actual peer is external and not a trusted proxy.
|
|
"""
|
|
from mjr_am_backend.routes.core import security as sec
|
|
|
|
with patch.object(sec, "_is_trusted_proxy", return_value=False):
|
|
result = sec._check_write_access(
|
|
peer_ip="203.0.113.99",
|
|
headers={"X-Forwarded-For": "127.0.0.1"},
|
|
)
|
|
assert not result.ok
|
|
assert result.code in {"FORBIDDEN", "AUTH_REQUIRED"}
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 6. Malformed metadata payloads — no crash, structured Err
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestMalformedMetadataPayloads:
|
|
HOSTILE_FILENAME_PAYLOADS = [
|
|
"",
|
|
"\x00",
|
|
"../../../etc/passwd",
|
|
"a" * 10_000, # oversized path string
|
|
"C:\\windows\\system32\\cmd.exe",
|
|
]
|
|
|
|
def test_safe_rel_path_never_raises(self):
|
|
from mjr_am_backend import path_utils
|
|
|
|
for payload in self.HOSTILE_FILENAME_PAYLOADS:
|
|
try:
|
|
result = path_utils.safe_rel_path(payload)
|
|
# Result is None, Path(""), or a safe relative path — never raises
|
|
assert result is None or isinstance(result, Path)
|
|
except Exception as exc:
|
|
pytest.fail(f"safe_rel_path({payload!r:.40}) raised: {exc}")
|
|
|
|
def test_normalize_path_never_raises(self):
|
|
from mjr_am_backend import path_utils
|
|
|
|
for payload in self.HOSTILE_FILENAME_PAYLOADS:
|
|
try:
|
|
result = path_utils.normalize_path(payload)
|
|
assert result is None or isinstance(result, Path)
|
|
except Exception as exc:
|
|
pytest.fail(f"normalize_path({payload!r:.40}) raised: {exc}")
|