131 lines
4.2 KiB
Python
131 lines
4.2 KiB
Python
"""Storage path handling, rate limiting, and response hardening."""
|
|
from __future__ import annotations
|
|
|
|
import pytest
|
|
|
|
from dsh_sync.config import Config
|
|
from dsh_sync.ratelimit import SlidingWindow
|
|
from dsh_sync.storage import artifact_path, artifact_relpath
|
|
|
|
|
|
# ------------------------------------------------------------------ storage
|
|
|
|
|
|
def _cfg(tmp_path) -> Config:
|
|
return Config(
|
|
data_dir=tmp_path,
|
|
plugins_dir=tmp_path / "plugins",
|
|
database_url="postgresql://unused",
|
|
bootstrap_admin_token=None,
|
|
invite_code=None,
|
|
admin_username="admin",
|
|
max_plugin_bytes=1024,
|
|
)
|
|
|
|
|
|
def test_artifact_relpath_is_relative(tmp_path):
|
|
"""Stored paths must not embed a machine-specific prefix."""
|
|
rel = artifact_relpath("demo", "1.0.0")
|
|
assert rel == "demo/1.0.0.tar.gz"
|
|
assert not rel.startswith("/") and ":\\" not in rel
|
|
|
|
|
|
def test_artifact_path_resolves_under_plugins_dir(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
p = artifact_path(cfg, "demo/1.0.0.tar.gz")
|
|
assert p == (tmp_path / "plugins" / "demo" / "1.0.0.tar.gz").resolve()
|
|
assert p.is_relative_to(cfg.plugins_dir.resolve())
|
|
|
|
|
|
def test_artifact_path_accepts_legacy_absolute_rows(tmp_path):
|
|
"""Rows written by older builds hold absolute paths; they must still load."""
|
|
cfg = _cfg(tmp_path)
|
|
absolute = tmp_path / "plugins" / "demo" / "2.0.0.tar.gz"
|
|
assert artifact_path(cfg, str(absolute)) == absolute
|
|
|
|
|
|
def test_artifact_path_rejects_escape(tmp_path):
|
|
cfg = _cfg(tmp_path)
|
|
with pytest.raises(ValueError):
|
|
artifact_path(cfg, "../../etc/passwd")
|
|
|
|
|
|
# ---------------------------------------------------------------- ratelimit
|
|
|
|
|
|
def test_sliding_window_allows_up_to_limit():
|
|
w = SlidingWindow(limit=3, window_seconds=60)
|
|
for _ in range(3):
|
|
w.check("1.2.3.4")
|
|
|
|
|
|
def test_sliding_window_blocks_over_limit():
|
|
from fastapi import HTTPException
|
|
|
|
w = SlidingWindow(limit=2, window_seconds=60)
|
|
w.check("k")
|
|
w.check("k")
|
|
with pytest.raises(HTTPException) as e:
|
|
w.check("k")
|
|
assert e.value.status_code == 429
|
|
assert "Retry-After" in e.value.headers
|
|
|
|
|
|
def test_sliding_window_is_per_key():
|
|
w = SlidingWindow(limit=1, window_seconds=60)
|
|
w.check("a")
|
|
w.check("b") # different peer, own budget
|
|
|
|
|
|
def test_sliding_window_expires():
|
|
import time
|
|
|
|
w = SlidingWindow(limit=1, window_seconds=0.05)
|
|
w.check("k")
|
|
time.sleep(0.08)
|
|
w.check("k") # window rolled over
|
|
|
|
|
|
def test_invite_endpoint_rate_limited(client):
|
|
"""Wrong invite codes stop being free after the budget is spent."""
|
|
codes = [client.post("/v1/tokens", json={"username": f"u{i}", "invite_code": "nope"})
|
|
for i in range(12)]
|
|
assert all(c.status_code == 403 for c in codes[:10]), [c.status_code for c in codes]
|
|
assert codes[10].status_code == 429
|
|
assert "Retry-After" in codes[10].headers
|
|
|
|
|
|
# ------------------------------------------------------------------ headers
|
|
|
|
|
|
def test_security_headers_on_api_responses(client):
|
|
h = client.get("/healthz").headers
|
|
assert h["x-content-type-options"] == "nosniff"
|
|
assert h["x-frame-options"] == "DENY"
|
|
assert h["referrer-policy"] == "no-referrer"
|
|
# camera must stay same-origin: the PWA's in-app scanner needs getUserMedia
|
|
assert "camera=(self)" in h["permissions-policy"]
|
|
|
|
|
|
def test_security_headers_on_pages(client):
|
|
for path in ("/app", "/apk"):
|
|
h = client.get(path).headers
|
|
assert h["x-content-type-options"] == "nosniff", path
|
|
assert h["x-frame-options"] == "DENY", path
|
|
|
|
|
|
def test_successful_redeem_clears_the_rate_limit(client):
|
|
"""A user who mistypes the code must not stay locked out after succeeding."""
|
|
for i in range(5):
|
|
assert client.post("/v1/tokens", json={"username": f"x{i}", "invite_code": "bad"}
|
|
).status_code == 403
|
|
ok = client.post("/v1/tokens", json={"username": "good", "invite_code": "invite-123"})
|
|
assert ok.status_code == 200, ok.text
|
|
|
|
from conftest import INVITE_CODE
|
|
|
|
# Bucket was cleared, so the next peer-budget starts fresh rather than
|
|
# inheriting the 5 failures.
|
|
again = client.post("/v1/tokens", json={"username": "good2", "invite_code": INVITE_CODE})
|
|
assert again.status_code == 200, again.text
|